[Searchable Snapshot] Design file caching mechanism for block based files

[andrross]

Currently searchable snapshots download Lucene files using a chunking approach to only download the data that is needed to service a query. It should use a node-level LRU cache that will use up to a configurable amount of local disk space to avoid re-downloading the same parts of frequently-accessed files. All shards on the node should share the same logical cache, meaning that if one shard is queried exclusively then it should use up to the entire cache space configured for the node.

Open questions:

• How is the cache size configured for a node? Is there a reasonable default if no configuration is provided?
• Where on disk should the data be cached? i.e. inside the same directory structure as the rest of the index data? Or are there use cases where the cache would want to be a dedicated disk or mount that would require a separate top-level directory?
• How should a node report cache statistics and/or utilization?

[aabukhalil]

I'll be working on this

[anasalkouz]

Are we building this with the assumption that the current searcher node is going to only be used for remote index? if no, then we need to think of some complex use-cases where the same node has local index (Hot) and remote index with cache. Which mean we need to build some mechanisms to control the storage usage for hot vs cached data.

[aabukhalil]

here is my findings for the caching mechanism
path.data setting is accepted as a list so the data node can be configured statically to use multiple data paths and/or volumes.

Cache Scope

the file system cache will be data node local cache. Single data node might have multiple data volumes, multiple indices and shards, and maybe multiple caches. We need to decide what Volume:Cache mapping would be (1:1, 1:M, ..) and Cache:Index mapping.

I think that the most flexible ultimate and highly configurable interface for defining caches is to define a static (maybe later dynamic API exposed) named cache in the yml file like:

remote_store_cache:
    cache_1_name:
	  path: /path/where/to/put/cache1
	  reserved_percent: 30
    cache_2_name:
	  path: /path/where/to/put/cache2
	  reserved_percent: 30

or can be flattened as (because map Setting is not supported)

named_caches: [cache_1_name, cache_2_name]
named_cache.cache_1_name.path:
named_cache.cache_1_name.reserved_percent:
named_cache.cache_2_name.path:
named_cache.cache_2_name.reserved_percent:

and when we restore a remote stored index modify the api /_snapshot/test_repository/test_snapshot/_restore to include cache_name

{
    "indices": "nyc_taxis",
    "rename_pattern": "(.+)",
    "rename_replacement": "restored_$1",
    "storage_type": "remote_snapshot",
    "cache_name": "named_defined_cache" // if not provided, nothing will be cached ?
}

This way users can decide how many caches needed, what volume is being used by each cache, whether to share cache size with local hot indices or not, which indices uses which cache and how fast and big each cache is. This interface basically will allow user to decide node role to be hot, cold, warm or mixed node. For now I don't see a need to support single cache backed by multiple volumes / data paths. Disadvantage to this approach, it's harder to define default behavior (when no cache is defined) especially with the size reservation logic.

Other options

• Cache per Index: will be stored at same data path of the index and more convenient to use and will have less configuration but it's hard to satisfy cases where a shared cache is needed or even when index name will not be known ahead of time or in future stages when index metadata will not exist on the node, it's better to have dedicated directory for the cache which will make data cleaning easier (more details below). Also this might have too overhead to maintain a LRU cache per index. Multiple caches might be able to share same quote from disk usage but it's complicated to implement with many open question.
• Single Cache per Node: might need to implement a cache backed by multiple data paths and it's less configurable and now all indices will use same cache. Complexity to implement multiple named caches is slightly more than this single cache so I think it's okay to implement the named cache at first iteration.

Cache Size and Reservation

Depending on the cache scope decision, a cache might be sharing space with hot Lucene indices data. Hot shards might grow, get relocated , ... so either we need to make the cache size dynamic (best effort to use not-utilized space by cache and hot shard will more priority to use the disk and shard relocation logic need to have access to the cache to make it shrink itself and will make it more complex ) or the cache size should be fixed size and reserved which I think it is more simple to reserve size and have more decoupled components.

For now, I think defining cache by a fixed percentage to use of the volume is easier to use and more dynamic for users than absolute bytes value.
When a cache is defined to be e.g. 30% of a volume or X bytes sized meaning that the cache might have size of [0-X) on disk at any time before it start evicting files and replacing them, the cache initially will not have these bytes occupied on disk.

Cache Disk reservation logic will impact all of:

• NodesStats as fs stats need to take into consideration the cache reserved size
• DiskThresholdDecider once monitor service having the reservation logic injected it should work out of the box
• MonitorService need to take into consideration the cache reserved size
• org.opensearch.cluster.ClusterInfo::reservedSpace
• org.opensearch.cluster.InternalClusterInfoService::buildShardLevelInfo
• org.opensearch.index.store.StoreStats##reservedSize
  so that
  org.opensearch.cluster.routing.allocation.DiskThresholdMonitor
  and
  org.opensearch.cluster.routing.allocation.decider.DiskThresholdDecider
  take into consideration the new disk reservation logic

Reservation Logic

When creating the cache, a validations are needed to make sure:

• (cache defined reserved size - current cache phantom files + lucene indices file size) <= cluster.routing.allocation.disk.watermark.high and cluster.routing.allocation.disk.watermark.low and so on
• When cache is initiated on node start procedure, it's needed to walk through cache dir files and count their sizes
• For now we don't need to handle logic if cache data path has changed and old path has ghost files

Default Behavior

Given that it's easier to implement fixed sized cache with reservation logic, and, depending on the cache scope decision not all data nodes might have a remote stored index, so it does not make sense to include a reserved file system cache by default because that would be waste of resources. To actually have default caching enabled, we need to implement dynamic sized cache.

So for default behavior these are options I think of:

• having a dynamic sized cache which might be complex ? if feasible then it can be enabled by default
• having a reserved fixed size cache by default which will not be suitable when users does not have remote store indices or don't need caching at all
• not having a cache by default, but users need to explicitly define caches. having a cache on node level will somehow introduce new node type which is cachable reader. it should be a reader node with an option to put a cache.

Stats to publish and how ? scope of the stats ?

since cache is scoped per node, we can publish stats on node level or named cache level. named cache level is more granular with the cost of more expensive stats to maintain and transmit and we do both. For it's easier to have stats per named cache.

Stats to publish:
org.opensearch.common.cache.Cache.CacheStats (hits, misses, evictions) + utilized bytes (cache all bytes content) + reserved size per named cache and publish it to nodes stats

org.opensearch.node.NodeService#stats -> should publish
org.opensearch.monitor.fs.FsService#stats() should contain the named caches stats or introduce new field in org.opensearch.action.admin.cluster.node.stats.NodeStats if does not make sense to include remote store cache in fs stats.

How to build LRU Cache ?

We should not build a cache from scratch for this. Any standard on-heap LRU cache with listeners and weighter should work to implement the file system caching. Basically the on-heap cache will hold Reference to file on disk as below

static class FileOnDisk {
        int sizeOnDisk;

        public FileOnDisk(int sizeOnDisk) {
            this.sizeOnDisk = sizeOnDisk;
        }
    }

  Cache<String, FileOnDisk> theCache =
            CacheBuilder.newBuilder()
                .maximumWeight(10) // max size in bytes
                .weigher((Weigher<String, FileOnDisk>) (key, value) -> value.sizeOnDisk)
                .build();

Looking at current LRU cache in our code base, look like org.opensearch.common.cache has the functionality we need and we can use it. But it looks like it's an out dated version of Guava Cache with a lot of locking to maintain the LRU. So as an improvement to this remote store feature and to any user of that cache maybe we need to upgrade to latest Guava Cache or it's successor Caffeine Cache which looks like being used by many data stores like Solr, Cassandra,Druid, ...

Minimum Viable Product

If we define a dedicated role for remote store reader (change hot shard allocation decider to exclude these nodes) then no need for cache size reservation logic at all since all data volumes will be exclusively used by cache. But this does not solve the concern of default out of the box behavior since used need to explicitly set node role to only remote reader.

So depending the effort, I think that implementing cache reservation logic is more flexible for users and better future investment

[aabukhalil]

Task Breakdown:

• ([Searchable Snapshot] Implement disk reservation logic to be used for local file system cache #5508) Given Cache max reserved size, current cache occupied size, Implement size reservation logic to be reflected on NodesStats::FileSystemStats, Deciders, Monitoring Service
• ([Searchable Snapshot] Implement FileSystem Tracking LRU Cache with RefCounting #4965) Implement FileSystem Tracking LRU Cache with RefCounting (Cache should not delete a file being used so when cloning IndexInput refCount needs ++ and when being closed -- the counter) on Top of existing Cache
• ([Searchable Snapshot] IndexInput clones close handling #5243) When Lucene clones IndexInput it will not close it, it will only close the original IndexInput. So we need to keep track of the cloned IndexInputs and close them when they are not used anymore. Scope here is find how to detect the unused IndexInput, is relying on GC and Cleaner is the only option here ? and then implement it
• ([Searchable Snapshot] Add Integ test to make there is no resource leak #5244) Searchable Snapshot and how IndexInput is being used, it has lot of clones, closings and many files handling logic. If we missing closing any file then we will cause resource leaks. We Integ test to make sure there is no resource leak. This is possible (theoretically) by creating remote backed index, do some searches, and then delete the index. And monitor the file descriptor held by process affecting the caching directory.
• Implement the proposed interface to define the named cache specs, basically transform new yml config to List<CacheDefintion> POJOs.
• ([Searchable Snapshot] Implement cache restore mechanism and track phantom files #5980) Right now, FileCache capacity is computed arbitrary using the first data path x 50% (hard coded) of total size. Also, for now, Cached files are stored within the corresponding indices directories. Also, for now, when node restart it won't put the cached files into the cache but will ignore them (phantom files). Next step is to decide which cache scope we should use and where to store the cached data and construct the cache accordingly. Change Node bootstrap to construct the named cached based on the List, it should include the cached files walk through logic as well to restore them to cache or to delete them.
• Change restore API to include and validate the cache name
• ([Searchable Snapshot] Add listener to Index events to update local file system cache entries #5981) Add listener to index events so that when remote index is deleted it's corresponding cached files are de-referrenced and deleted from cache
• ([Searchable Snapshot] Expose local file system cache stats into node stats #5982) Expose all file system cache stats (usage, activeUsage, hitCount, missCount, ...) into NodeStats API. Stats already exist as POJOs and computed correctly but we need to include them in node stats.
• For now, OnDemandVirtualFileSnapshotIndexInput is never cached or stored into disk but it will always use ByteArrayIndexInput since whole IndexInput is included in the hash in memory. Make sure this does not have any performance penalty
• ([Searchable Remote Index] Add a REST API for clearing file cache #6030)
• ([Searchable Snapshot] Revist using other open source LRUCache implementation for File System caching  #6225)
• ([Searchable Snapshot] Assess File System caching impact on heap memory  #6227)

[aabukhalil]

Open Questions

• Since the cache should not be able to de-reference and delete a file while being used by any IndexInput (since there might be many IndexInputs created and backed by same file). What will happen if the cache is having high throughput concurrent searches where current held files exceed the cache capacity and IndexInput is needed to serve a request ? what if reaching disk size limits ? should these requests be blocked till there is more room to get cached ? if yes, how to prevent dead locks where multiple search requests are holding partial IndexInputs and all wait to empty room. Should we keep current implementation of no caching and using the ByteBufferIndexInput and when cache is full just use it ? what Circuit Breaker mechanism can we use here ?
• If we will be implementing the named cache mechanism, how to handle situation where multiple data nodes having inconsistent cache names ? should coordinator re-route queries based on cache name ? if yes then cache name would act as a role itself. or should the cluster fail if it detect inconsistent static settings between nodes ? or just leave it as user responsibility ?

[andrross]

Thanks @aabukhalil ! I'm still digging into a lot of this, but just wanted to point out an existing issue that documents the poor performance of the existing cache in OpenSearch.