Implementing Auto-Refreshing Caches

Flyte provides a thread-safe, general-purpose auto-refreshing cache in the flytestdlib.autorefreshcache package. This cache is designed for scenarios where you need to track the status of external resources (like jobs in a remote service) asynchronously without blocking your main execution flow.

In this tutorial, you will build a cache that tracks the status of external jobs, automatically updating their state in the background until they reach a terminal state.

Prerequisites

To follow this tutorial, you need the following packages imported:

import (
    "context"
    "time"

    "github.com/flyteorg/flyte/v2/flytestdlib/autorefreshcache"
    "github.com/flyteorg/flyte/v2/flytestdlib/promutils"
    "k8s.io/client-go/util/workqueue"
)

Step 1: Define the Cache Item

Every item stored in the cache must implement the autorefreshcache.Item interface. This interface requires an IsTerminal() method, which the cache uses to determine if it should stop refreshing the item.

Create a struct to represent your job status:

type JobStatus string

const (
    StatusStarted   JobStatus = "Started"
    StatusSucceeded JobStatus = "Succeeded"
)

type JobItem struct {
    id     string
    status JobStatus
}

// IsTerminal returns true if the item should no longer be refreshed.
// Once a job succeeds, we stop polling the external service.
func (j *JobItem) IsTerminal() bool {
    return j.status == StatusSucceeded
}

func (j *JobItem) ID() string {
    return j.id
}

Step 2: Implement the Sync Logic

The SyncFunc is the core of the auto-refresh mechanism. It is responsible for fetching the latest state of a batch of items from your external source.

func syncJobStatus(ctx context.Context, batch autorefreshcache.Batch) ([]autorefreshcache.ItemSyncResponse, error) {
    updatedItems := make([]autorefreshcache.ItemSyncResponse, 0, len(batch))
    
    for _, obj := range batch {
        // Retrieve the current item from the wrapper
        oldItem := obj.GetItem().(*JobItem)
        
        // Simulate a call to an external service to get the latest status
        // In a real scenario, you would call your API here
        newStatus := StatusSucceeded 
        
        if newStatus != oldItem.status {
            updatedItems = append(updatedItems, autorefreshcache.ItemSyncResponse{
                ID:     oldItem.ID(),
                Item:   &JobItem{id: oldItem.id, status: newStatus},
                Action: autorefreshcache.Update,
            })
        }
    }
    
    return updatedItems, nil
}

The SyncFunc returns a list of ItemSyncResponse objects. If an item has changed, you return the Update action. If nothing changed, you can omit it from the response or return Unchanged.

Step 3: Initialize the Cache

Use NewAutoRefreshCache to create the cache instance. You need to provide a name for metrics, the sync callback, a rate limiter, and the resync interval.

func CreateCache(scope promutils.Scope) (autorefreshcache.AutoRefresh, error) {
    // The rate limiter controls how many sync operations happen concurrently
    rateLimiter := workqueue.DefaultTypedControllerRateLimiter[*autorefreshcache.Batch]()
    
    // The resyncPeriod defines how often the cache attempts to refresh non-terminal items
    resyncPeriod := 30 * time.Second
    
    // parallelizm: number of workers processing the sync queue
    // size: maximum number of items in the LRU cache
    cache, err := autorefreshcache.NewAutoRefreshCache(
        "job-tracker",
        syncJobStatus,
        rateLimiter,
        resyncPeriod,
        10,  // parallelizm
        100, // size
        scope,
    )
    
    return cache, err
}

Step 4: Start and Use the Cache

The cache does not start refreshing until you call Start(). You should pass a context that you can cancel to shut down the background workers gracefully.

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    scope := promutils.NewScope("my_app")
    cache, _ := CreateCache(scope)

    // Start the background refresh loop
    err := cache.Start(ctx)
    if err != nil {
        panic(err)
    }

    // Add an item to the cache
    job := &JobItem{id: "job-1", status: StatusStarted}
    _, err = cache.GetOrCreate(job.id, job)

    // Retrieve the item later
    item, err := cache.Get("job-1")
    if err == nil {
        currentJob := item.(*JobItem)
        fmt.Printf("Job %s status: %s\n", currentJob.id, currentJob.status)
    }
}

Step 5: Handling Deletions

If you no longer need to track an item, use DeleteDelayed. Note that Flyte's autoRefresh implementation does not delete items immediately. Instead, it queues them for deletion in the next sync cycle.

// The item will be removed from the cache during the next background sync
err := cache.DeleteDelayed("job-1")

Until the next sync cycle runs, Get and GetOrCreate will continue to return the item in its last known state.

Monitoring and Metrics

The autoRefresh cache automatically exports several Prometheus metrics via the provided promutils.Scope:

• sync_errors: Counter for errors encountered during the sync process.
• lru_evictions: Counter for items evicted from the LRU cache due to size limits.
• latency: Summary of the time taken for sync operations.
• cache_hit / cache_miss: Counters for Get operations.
• size: Gauge representing the current number of items in the cache.

These metrics are prefixed with the name you provided during initialization (e.g., job-tracker).