Struct PipelineCache 

pub(crate) struct PipelineCache {
    pipelines: Cache<String, Arc<Pipeline>>,
    original_pipelines: Cache<String, PipelineContent>,
    failover_cache: Cache<String, PipelineContent>,
}

Pipeline cache is located on a separate file on purpose, to encapsulate inner cache. Only public methods are exposed.

pipelines and original_pipelines are keyed by the requested schema so a lookup is a single key probe, as [Cache::try_get_with] requires; resolving it to a stored schema is the loader’s job. failover_cache has no loader and keeps the stored-schema key.

Fields

pipelines: Cache<String, Arc<Pipeline>>

original_pipelines: Cache<String, PipelineContent>

failover_cache: Cache<String, PipelineContent>

If the pipeline table is invalid, we can use this cache to prevent failures when writing logs through the pipeline The failover cache never expires, but it will be updated when the pipelines cache is updated.

Implementations

impl PipelineCache

pub(crate) fn new(ttl: Duration) -> Self

pub(crate) async fn get_pipeline_with( &self, schema: &str, name: &str, version: PipelineVersion, init: impl Future<Output = Result<Arc<Pipeline>>>, ) -> Result<Arc<Pipeline>>

Concurrent misses on the same key share one init call.

pub(crate) async fn get_pipeline_str_with( &self, schema: &str, name: &str, version: PipelineVersion, init: impl Future<Output = Result<PipelineContent>>, ) -> Result<PipelineContent>

Concurrent misses on the same key share one init call.

pub(crate) async fn get_failover_cache( &self, schema: &str, name: &str, version: PipelineVersion, ) -> Result<Option<PipelineContent>>

Resolves across schemas, unlike the loaded caches: a pipeline stored under the empty schema is reachable from any schema.

pub(crate) async fn insert_failover_cache( &self, content: PipelineContent, with_latest: bool, )

pub(crate) async fn on_pipeline_created(&self, content: PipelineContent)

Dropping the stale latest aliases also clears the failover entries, so the new version is written back: an outage before the first read-back would otherwise have nothing to fall back on.

pub(crate) async fn invalidate(&self, name: &str, version: PipelineVersion)

Sweeps every schema and all three caches: the latest alias always, plus version when given.