[processor/spanmetrics] Fix getting key from cache error

[Frapschen]

Description:
I get an error in my app when it sent span to a otel collector which configed a spanmetrics processor with defaultDimensionsCacheSize:

2022/10/27 02:37:46 rpc error: code = Unknown desc = value not found in metricKeyToDimensions cache by key "amamba\x00amamba.io.api.pipeline.v1alpha1.Pipelines/ReplayPipelineRun\x00SPAN_KIND_SERVER\x00STATUS_CODE_OK\x00cd7b102e-fbc5-4556-a0fd-718298df3de9\x00amamba-system\x00demo-dev-worker-03\x00amamba-apiserver-66c9486b55-5l4n8"

I trick the error to:

opentelemetry-collector-contrib/processor/spanmetricsprocessor/processor.go

Lines 335 to 344 in 92ad54f

	func (p *processorImp) getDimensionsByMetricKey(k metricKey) (*pcommon.Map, error) {
	if item, ok := p.metricKeyToDimensions.Get(k); ok {
	if attributeMap, ok := item.(pcommon.Map); ok {
	return &attributeMap, nil
	}
	return nil, fmt.Errorf("type assertion of metricKeyToDimensions attributes failed, the key is %q", k)
	}
	return nil, fmt.Errorf("value not found in metricKeyToDimensions cache by key %q", k)
	}

I think the error will occur when the added keys are more than processorImp.defaultDimensionsCacheSize however the processorImp.callSum doesn't have any limitation, it contains the key and the key's value. I add a patch to get key from the metricKeyToDimensions.EvictedItems to handle this scene.

[TylerHelmuth]

pinging @albertteoh as code owner

[albertteoh]

I thought metricKeyToDimensions .Get(...) already does this?

opentelemetry-collector-contrib/processor/spanmetricsprocessor/internal/cache/cache.go

Line 60 in 0ea5dc4

val, ok := c.evictedItems[key]

[Frapschen]

yes, it do use the evictedItems , but still can't not explain why this error occur

[Frapschen]

I debug for a while, I catch the point this time that:

opentelemetry-collector-contrib/processor/spanmetricsprocessor/processor.go

Lines 313 to 324 in 9810bb0

	for key := range p.callSum {
	mCalls := ilm.Metrics().AppendEmpty()
	mCalls.SetName("calls_total")
	mCalls.SetEmptySum().SetIsMonotonic(true)
	mCalls.Sum().SetAggregationTemporality(p.config.GetAggregationTemporality())
	dpCalls := mCalls.Sum().DataPoints().AppendEmpty()
	dpCalls.SetStartTimestamp(pcommon.NewTimestampFromTime(p.startTime))
	dpCalls.SetTimestamp(pcommon.NewTimestampFromTime(time.Now()))
	dpCalls.SetIntValue(p.callSum[key])
	dimensions, err := p.getDimensionsByMetricKey(key)

It loops p.callSum containing all key-values since the collector is started to find a key in the cache, but the cache only has 1000 items by default. If the built different key is large than 1000, then the next loop of p.callSum will raise an error.

[albertteoh]

Ah yes, I think you're right about the fact that p.callSum accumulates keys indefinitely since startup, unless if the user has configured delta temporality.

For cumulative temporality, while I think we should still keep all keys since collector startup (so the counts are correct), I don't think we should loop over this entire set of metric keys. Instead we should only loop over those that relate to the current batch of spans received, by somehow marking a key as "dirty" if it relates to a span's set of metric keys.

What do you think?

I can put together some tests locally to test this theory.

[Frapschen]

I think so, we just loop over those keys that relate to the current batch of spans received. Do I need close this PR and wait for your PR or I fix it in this PR?

[albertteoh]

Please go ahead with the fix 👍🏼

Note that #15710 involves a fairly major refactor of the spanmetrics processor, so I suggest to wait for #15710 to be merged first.

As it's quite a large refactor, it might make more sense to close this PR and create a new one based off the refactored version.

[albertteoh]

@Frapschen FYI #15710 has been merged.