Method Query

runtime/query.go:70–157

(ctx context.Context, instanceID string, query Query, priority int)

}

func (r *Runtime) Query(ctx context.Context, instanceID string, query Query, priority int) error {
	qk := query.Key()
	// If key is empty, skip caching
	if qk == "" {
		return query.Resolve(ctx, r, instanceID, priority)
	}

	// Get dependency cache keys and optionally the underlying OLAP connector
	ctrl, err := r.Controller(ctx, instanceID)
	if err != nil {
		return err
	}
	deps := query.Deps()
	depKeys := make([]string, 0, len(deps))
	for _, dep := range deps {
		// Get the dependency resource
		res, err := ctrl.Get(ctx, dep, false)
		if err != nil {
			// Deps are approximate, not exact (see docstring for Deps()), so they may not all exist
			continue
		}

		// Add to cache key.
		// Using StateUpdatedOn instead of StateVersion because the state version is reset when the resource is deleted and recreated.
		key := fmt.Sprintf("%s:%s:%d:%d", res.Meta.Name.Kind, res.Meta.Name.Name, res.Meta.StateUpdatedOn.Seconds, res.Meta.StateUpdatedOn.Nanos/int32(time.Millisecond))
		if mv := res.GetMetricsView(); mv != nil {
			cacheKey, ok, err := r.metricsViewCacheKey(ctx, instanceID, res.Meta.Name.Name, priority)
			if err != nil {
				return err
			}
			if !ok {
				// skip caching
				return query.Resolve(ctx, r, instanceID, priority)
			}
			key = key + ":" + string(cacheKey)
		}
		depKeys = append(depKeys, key)
	}

	// If there were no known dependencies, skip caching
	if len(depKeys) == 0 {
		return query.Resolve(ctx, r, instanceID, priority)
	}

	// Build cache key
	depKey := strings.Join(depKeys, ";")
	key := queryCacheKey{
		instanceID:    instanceID,
		queryKey:      qk,
		dependencyKey: depKey,
	}.String()

	// Try to get from cache
	if val, ok := r.queryCache.cache.Get(key); ok {
		observability.AddRequestAttributes(ctx, attribute.Bool("query.cache_hit", true))
		return query.UnmarshalResult(val)
	}
	observability.AddRequestAttributes(ctx, attribute.Bool("query.cache_hit", false))