diff --git a/pkg/transaction/metric/cachestorage.go b/pkg/transaction/metric/cachestorage.go index 60d604f02..89c540ccc 100644 --- a/pkg/transaction/metric/cachestorage.go +++ b/pkg/transaction/metric/cachestorage.go @@ -172,8 +172,9 @@ func (c *cacheStorage) loadMetrics(storageCache cache.Cache) { appDetails.ConsumerOrgID = cm.App.ConsumerOrgID } + var metric *centralMetric if cm.Unit != nil { - c.collector.AddCustomMetricDetail(models.CustomMetricDetail{ + metric = c.collector.updateCustomMetric(models.CustomMetricDetail{ APIDetails: apiDetails, AppDetails: appDetails, UnitDetails: models.Unit{ @@ -181,6 +182,7 @@ func (c *cacheStorage) loadMetrics(storageCache cache.Cache) { }, Count: cm.Count, }) + c.rekeyLoadedMetric(storageCache, metric, cacheKey, cm) continue } @@ -188,7 +190,6 @@ func (c *cacheStorage) loadMetrics(storageCache cache.Cache) { continue } - var metric *centralMetric if len(cm.Values) > 0 { // legacy cache written before the Min/Max/Avg counter, replay each // raw duration sample into the new counter one at a time @@ -208,17 +209,25 @@ func (c *cacheStorage) loadMetrics(storageCache cache.Cache) { }, cm.Count, cm.Min, cm.Max, cm.Avg) } - newKey := metric.getKey() - if newKey != cacheKey { - c.storageLock.Lock() - storageCache.Delete(cacheKey) - c.storageLock.Unlock() - } - storageCache.Set(newKey, cm) + c.rekeyLoadedMetric(storageCache, metric, cacheKey, cm) } } } +// rekeyLoadedMetric update the metric key, not serialized, on load from cache +func (c *cacheStorage) rekeyLoadedMetric(storageCache cache.Cache, metric *centralMetric, cacheKey string, cm cachedMetric) { + if metric == nil { + return + } + newKey := metric.storageKey() + if newKey != cacheKey { + c.storageLock.Lock() + storageCache.Delete(cacheKey) + c.storageLock.Unlock() + } + storageCache.Set(newKey, cm) +} + func (c *cacheStorage) updateMetric(cached cachedMetricInterface, metric *centralMetric) { if !c.isInitialized { return @@ -227,7 +236,7 @@ func (c *cacheStorage) updateMetric(cached cachedMetricInterface, metric *centra c.storageLock.Lock() defer c.storageLock.Unlock() - c.storage.Set(metric.getKey(), metric.createCachedMetric(cached)) + c.storage.Set(metric.storageKey(), metric.createCachedMetric(cached)) } func (c *cacheStorage) removeMetric(metric *centralMetric) { @@ -238,7 +247,7 @@ func (c *cacheStorage) removeMetric(metric *centralMetric) { c.storageLock.Lock() defer c.storageLock.Unlock() - c.storage.Delete(metric.getKey()) + c.storage.Delete(metric.storageKey()) } func (c *cacheStorage) save() { diff --git a/pkg/transaction/metric/centralmetric.go b/pkg/transaction/metric/centralmetric.go index c9f0d2f79..e3d3446cf 100644 --- a/pkg/transaction/metric/centralmetric.go +++ b/pkg/transaction/metric/centralmetric.go @@ -215,6 +215,10 @@ type centralMetric struct { Reporter *Reporter `json:"reporter,omitempty"` Observation *models.ObservationDetails `json:"-"` EventID string `json:"-"` + key string `json:"-"` + + // used as part of the key to separate current from new metrics + groupStartTime int64 // ctx is the metric context reported when the agent added the data to the collector ctx transactionContext @@ -315,6 +319,10 @@ func (a *centralMetric) addTransactionFields(fields logrus.Fields) logrus.Fields // getKey - returns the cache key for the metric func (a *centralMetric) getKey() string { + if a.key != "" { + return a.key + } + appKey := unknown if a.ctx.AppDetails.ID != "" { appKey = sanitizeKeySegment(a.ctx.AppDetails.ID) @@ -336,7 +344,12 @@ func (a *centralMetric) getKey() string { } } - return strings.Join([]string{metricKeyPrefix, appKey, apiID, uniqueKey}, ".") + a.key = strings.Join([]string{metricKeyPrefix, appKey, apiID, uniqueKey}, ".") + return a.key +} + +func (a *centralMetric) storageKey() string { + return fmt.Sprintf("%s.%d", a.getKey(), a.groupStartTime) } func (a *centralMetric) createCachedMetric(cached cachedMetricInterface) cachedMetric { diff --git a/pkg/transaction/metric/metricscollector.go b/pkg/transaction/metric/metricscollector.go index e990da384..e73a84bfe 100644 --- a/pkg/transaction/metric/metricscollector.go +++ b/pkg/transaction/metric/metricscollector.go @@ -305,6 +305,10 @@ func (c *collector) AddCustomMetricDetail(detail models.CustomMetricDetail) { c.batchLock.Lock() defer c.batchLock.Unlock() + c.updateCustomMetric(detail) +} + +func (c *collector) updateCustomMetric(detail models.CustomMetricDetail) *centralMetric { logger := c.logger.WithField("handler", "customMetric"). WithField("apiID", detail.APIDetails.ID). WithField("appID", detail.AppDetails.ID). @@ -312,17 +316,17 @@ func (c *collector) AddCustomMetricDetail(detail models.CustomMetricDetail) { if detail.APIDetails.ID == "" { logger.Error("custom units require API information") - return + return nil } if detail.AppDetails.ID == "" { logger.Error("custom units require App information") - return + return nil } if detail.UnitDetails.Name == "" { logger.Error("custom units require Unit information") - return + return nil } logger.WithField("count", detail.Count).Debug("received custom unit report") @@ -346,7 +350,7 @@ func (c *collector) AddCustomMetricDetail(detail models.CustomMetricDetail) { counter := c.getOrRegisterGroupedCounter(metric.getKey()) counter.Inc(detail.Count) - c.updateMetricWithCachedMetric(metric, newCustomCounter(counter)) + return c.updateMetricWithCachedMetric(metric, newCustomCounter(counter)) } // AddAPIMetric - add api metric for API transaction, merging its counts and response stats into @@ -550,6 +554,7 @@ func (c *collector) updateMetricWithCachedMetric(metric *centralMetric, cached c groupKey, uniqueKey := splitMetricKey(metric.getKey()) groupedMetric := c.getOrRegisterGroupedMetrics(c.groupKeyWithStartTime(groupKey)) + metric.groupStartTime = c.metricStartTime.UnixMilli() // first api metric for sub+app+api+statuscode wins and becomes the template used for reporting metric = groupedMetric.getOrSetMetric(uniqueKey, metric) @@ -937,10 +942,12 @@ func (c *collector) processMetric(metricName string, groupedMetricInterface inte logger := c.logger. WithField("applicationID", desanitizeKeySegment(elements[1])). WithField("apiID", desanitizeKeySegment(elements[2])) - c.handleGroupedMetric(logger, groupedMetric, publishStartTime, metricName) + + // use the start time in the group + c.handleGroupedMetric(logger, groupedMetric, time.UnixMilli(groupStartTime), metricName) } -func (c *collector) handleGroupedMetric(logger log.FieldLogger, groupedMetric groupedMetrics, publishStartTime time.Time, registryKey string) { +func (c *collector) handleGroupedMetric(logger log.FieldLogger, groupedMetric groupedMetrics, startTime time.Time, registryKey string) { countersAdded := false // handle each api counter, on the first one add the counter information for k, apiCtr := range groupedMetric.apiCounters { @@ -957,7 +964,7 @@ func (c *collector) handleGroupedMetric(logger log.FieldLogger, groupedMetric gr counters = groupedMetric.counters countersAdded = true } - c.generateMetricEvent(counters, metric, publishStartTime, registryKey, groupedMetric) + c.generateMetricEvent(counters, metric, startTime, registryKey, groupedMetric) } // create metric with just custom units @@ -973,7 +980,7 @@ func (c *collector) handleGroupedMetric(logger log.FieldLogger, groupedMetric gr return } c.setMetricCounters(logger, metric, groupedMetric) - c.generateMetricEvent(groupedMetric.counters, metric, publishStartTime, registryKey, groupedMetric) + c.generateMetricEvent(groupedMetric.counters, metric, startTime, registryKey, groupedMetric) } } @@ -1015,13 +1022,13 @@ func (c *collector) setMetricsFromAPICounter(m *centralMetric, apiCtr *apiCounte } } -func (c *collector) generateMetricEvent(counters map[string]*counter, metric *centralMetric, publishStartTime time.Time, registryKey string, group groupedMetrics) { +func (c *collector) generateMetricEvent(counters map[string]*counter, metric *centralMetric, startTime time.Time, registryKey string, group groupedMetrics) { if metric.Units != nil && metric.Units.Transactions != nil && metric.Units.Transactions.Count == 0 { c.logger.Trace("skipping registry entry with no reported quantity") return } metric.Observation = &models.ObservationDetails{ - Start: util.ConvertTimeToMillis(publishStartTime), + Start: util.ConvertTimeToMillis(startTime), End: util.ConvertTimeToMillis(c.metricEndTime), } metric.Reporter = &Reporter{ @@ -1033,7 +1040,7 @@ func (c *collector) generateMetricEvent(counters map[string]*counter, metric *ce } // Generate app subscription metric - c.generateV4Event(counters, metric, publishStartTime, registryKey, group) + c.generateV4Event(counters, metric, startTime, registryKey, group) } func (c *collector) createV4Event(startTime int64, v4data V4Data) V4Event { @@ -1050,8 +1057,8 @@ func (c *collector) createV4Event(startTime int64, v4data V4Data) V4Event { } } -func (c *collector) generateV4Event(counters map[string]*counter, v4data V4Data, publishStartTime time.Time, registryKey string, group groupedMetrics) { - generatedEvent := c.createV4Event(publishStartTime.UnixMilli(), v4data) +func (c *collector) generateV4Event(counters map[string]*counter, v4data V4Data, startTime time.Time, registryKey string, group groupedMetrics) { + generatedEvent := c.createV4Event(startTime.UnixMilli(), v4data) c.metricLogger.WithFields(generatedEvent.getLogFields()).Info("generated") AddCondorMetricEventToBatch(generatedEvent, c.metricBatch, registryKey, counters, group) } @@ -1156,15 +1163,13 @@ func (c *collector) logMetric(msg string, metric *centralMetric) { // of being lost. Once every entry in the group has been acked, the group itself is removed from the // registry. func (c *collector) cleanupMetricCounters(registryKey string, counters map[string]*counter, group groupedMetrics, metric *centralMetric) { - c.storage.removeMetric(metric) - + // clean all counters and metrics _, statusKey := splitMetricKey(metric.getKey()) + c.removeStoredMetric(group, statusKey) empty := group.removeAndCheckEmpty(statusKey) for k := range counters { - if m, ok := group.getMetric(k); ok { - c.storage.removeMetric(m) - } + c.removeStoredMetric(group, k) empty = group.removeAndCheckEmpty(k) } @@ -1179,6 +1184,12 @@ func (c *collector) cleanupMetricCounters(registryKey string, counters map[strin Info("Published metrics report for API") } +func (c *collector) removeStoredMetric(group groupedMetrics, key string) { + if m, ok := group.getMetric(key); ok { + c.storage.removeMetric(m) + } +} + func GetStatusText(statusCode string) string { return sampling.GetStatusFromCodeString(statusCode).String() } diff --git a/pkg/transaction/metric/metricscollector_test.go b/pkg/transaction/metric/metricscollector_test.go index 132b71581..5d9ee4d9e 100644 --- a/pkg/transaction/metric/metricscollector_test.go +++ b/pkg/transaction/metric/metricscollector_test.go @@ -846,6 +846,316 @@ func TestMetricCollectorPublishesAllSubscriptionsAndCleansRegistry(t *testing.T) s.resetConfig() } +func metricStorageKeys(c *collector) []string { + cs := c.storage.(*cacheStorage) + var keys []string + for _, k := range cs.storage.GetKeys() { + if strings.HasPrefix(k, metricKeyPrefix+".") { + keys = append(keys, k) + } + } + return keys +} + +func metricRegistryGroups(c *collector) []string { + var names []string + c.registry.Each(func(name string, _ interface{}) { + if strings.HasPrefix(name, metricKeyPrefix+".") { + names = append(names, name) + } + }) + return names +} + +// TestMetricCollectorPublishesAllStatusesAndUnitsAndCleansStorage loads several HTTP status metrics and +// custom unit metrics, publishes them, and verifies every status/unit is present on the published events +// and that both the registry and the persisted cache storage are fully cleaned once the events are acked. +func TestMetricCollectorPublishesAllStatusesAndUnitsAndCleansStorage(t *testing.T) { + defer cleanUpCachedMetricFile() + s := &testHTTPServer{} + defer s.closeServer() + s.startServer() + traceability.SetDataDirPath(".") + + metricCollector, _ := setupMetricCollectorTest(t, s) + traceStatus = healthcheck.OK + runTestHealthcheck() + + // start from a clean registry and an empty in-memory storage so the assertions below observe only + // the metrics recorded by this test + metricCollector.registry = newRegistry() + freshStorage := newStorageCache(metricCollector).(*cacheStorage) + freshStorage.isInitialized = true + metricCollector.storage = freshStorage + + testClient := setupMockClient(0) + + // several HTTP statuses for one app/api - each distinct status text (Success/Failure/Exception) is + // tracked as its own metric entry within the app/api's registry group + for _, status := range []string{"200", "400", "500"} { + metricCollector.AddMetricDetail(Detail{ + APIDetails: apiDetails1, + StatusCode: status, + Duration: 10, + Bytes: 10, + AppDetails: models.AppDetails{ID: "app-1", Name: testManagedApp1}, + }) + } + + // custom units for a second app/api - these land in a separate registry group and publish together + // on a single custom-unit event + for _, unit := range []string{"widgets", "gadgets"} { + metricCollector.AddCustomMetricDetail(models.CustomMetricDetail{ + APIDetails: models.APIDetails{ID: "111", Name: "111"}, + AppDetails: models.AppDetails{ID: "app-2", Name: testManagedApp2}, + UnitDetails: models.Unit{Name: unit}, + Count: 3, + }) + } + + // before publishing: 3 status + 2 unit entries are cached, spread across 2 registry groups + assert.Len(t, metricStorageKeys(metricCollector), 5) + assert.Len(t, metricRegistryGroups(metricCollector), 2) + + assert.NoError(t, metricCollector.Execute()) + metricCollector.usagePublisher.Execute() + + mock := testClient.(*MockClient) + + // the batch should carry one event per status plus one custom-unit event + assert.Equal(t, 4, mock.eventsAcked) + assert.Len(t, mock.capturedEvents, 4) + + publishedStatuses := map[string]bool{} + publishedUnits := map[string]bool{} + for _, event := range mock.capturedEvents { + if metric := getMetricFromEvent(event); assert.NotNil(t, metric) && + metric.Units != nil && metric.Units.Transactions != nil && metric.Units.Transactions.Status != "" { + publishedStatuses[metric.Units.Transactions.Status] = true + } + // custom units are dropped when reconstructing from the event (json:"-"), so read them from the + // raw published data instead + if data := getRawEventData(event); data != nil { + if units, ok := data["units"].(map[string]interface{}); ok { + for name, v := range units { + if name != "transactions" && v != nil { + publishedUnits[name] = true + } + } + } + } + } + + assert.Equal(t, map[string]bool{"Success": true, "Failure": true, "Exception": true}, publishedStatuses) + assert.Equal(t, map[string]bool{"widgets": true, "gadgets": true}, publishedUnits) + + // once every event is acked, the registry and the persisted cache storage should both be free of + // metric entries + assert.Empty(t, metricRegistryGroups(metricCollector), "registry should be cleaned of all metric groups") + assert.Empty(t, metricStorageKeys(metricCollector), "cache storage should be cleaned of all metric entries") + + s.resetConfig() +} + +// TestMetricStorageKeysAreIsolatedByGenerationStartTime verifies that metrics for the same +// app/api/status collected in different generations are stored under distinct, start-time-qualified +// keys, so cleaning up a generation whose event was published does not remove a later generation's +// metric that was collected while the first was still in flight. +func TestMetricStorageKeysAreIsolatedByGenerationStartTime(t *testing.T) { + defer cleanUpCachedMetricFile() + s := &testHTTPServer{} + defer s.closeServer() + s.startServer() + traceability.SetDataDirPath(".") + + metricCollector, _ := setupMetricCollectorTest(t, s) + traceStatus = healthcheck.OK + runTestHealthcheck() + + metricCollector.registry = newRegistry() + freshStorage := newStorageCache(metricCollector).(*cacheStorage) + freshStorage.isInitialized = true + metricCollector.storage = freshStorage + + addDetail := func() { + metricCollector.AddMetricDetail(Detail{ + APIDetails: apiDetails1, + StatusCode: "200", + Duration: 10, + Bytes: 10, + AppDetails: models.AppDetails{ID: "app-1", Name: testManagedApp1}, + }) + } + + // generation 1 + metricCollector.metricStartTime = time.UnixMilli(60000) + addDetail() + + // generation 2 for the same app/api/status, as if collected while generation 1 is publishing + metricCollector.metricStartTime = time.UnixMilli(120000) + addDetail() + + // each generation is stored under its own start-time-qualified key and its own registry group + assert.Len(t, metricStorageKeys(metricCollector), 2) + assert.Len(t, metricRegistryGroups(metricCollector), 2) + + // find generation 1's registry group and clean it up, as happens when its published event is acked + var gen1Name string + var gen1Group groupedMetrics + metricCollector.registry.Each(func(name string, v interface{}) { + if strings.HasSuffix(name, ".60000") { + gen1Name = name + gen1Group, _ = v.(groupedMetrics) + } + }) + if !assert.NotEmpty(t, gen1Name) { + return + } + gen1Metric, ok := gen1Group.getMetric("Success") + if !assert.True(t, ok) { + return + } + metricCollector.cleanupMetricCounters(gen1Name, gen1Group.counters, gen1Group, gen1Metric) + + // generation 1 is gone from both registry and storage; generation 2 is untouched + if remaining := metricStorageKeys(metricCollector); assert.Len(t, remaining, 1) { + assert.True(t, strings.HasSuffix(remaining[0], ".120000"), + "generation 2 metric should remain in storage, got %s", remaining[0]) + } + if groups := metricRegistryGroups(metricCollector); assert.Len(t, groups, 1) { + assert.True(t, strings.HasSuffix(groups[0], ".120000"), + "generation 2 group should remain in registry, got %s", groups[0]) + } + + s.resetConfig() +} + +// TestMetricCacheRoundTripRekeysAndCleansUpWithoutOrphaning saves transaction metrics from two different +// generations, reloads them into a fresh collector, and verifies each entry is re-keyed to the current +// run's storage key (the key is recomputed from the reloaded context, not restored from a persisted +// cachedMetric.Key) and is fully removed from both the registry and storage on publish - i.e. no stale, +// orphaned cache entry survives the round trip. +func TestMetricCacheRoundTripRekeysAndCleansUpWithoutOrphaning(t *testing.T) { + cleanUpCachedMetricFile() + defer cleanUpCachedMetricFile() + s := &testHTTPServer{} + defer s.closeServer() + s.startServer() + traceability.SetDataDirPath(".") + + collector1, _ := setupMetricCollectorTest(t, s) + traceStatus = healthcheck.OK + runTestHealthcheck() + + // generation 1: a success transaction, keyed with start time 60000 + collector1.metricStartTime = time.UnixMilli(60000) + collector1.AddMetricDetail(Detail{ + APIDetails: apiDetails1, + StatusCode: "200", + Duration: 10, + Bytes: 10, + AppDetails: models.AppDetails{ID: "app-1", Name: testManagedApp1}, + }) + + // generation 2: a failure transaction, keyed with start time 120000 - this later value is what gets + // persisted as the metric start time and loaded by the next collector + collector1.metricStartTime = time.UnixMilli(120000) + collector1.AddMetricDetail(Detail{ + APIDetails: apiDetails1, + StatusCode: "400", + Duration: 10, + Bytes: 10, + AppDetails: models.AppDetails{ID: "app-1", Name: testManagedApp1}, + }) + + assert.Len(t, metricStorageKeys(collector1), 2) + collector1.storage.save() + + // reload into a fresh collector + collector2 := createMetricCollector().(*collector) + + // both metrics are restored and re-keyed to this run's start time (120000); the stale generation-1 + // key (60000) must not survive + reloadedKeys := metricStorageKeys(collector2) + assert.Len(t, reloadedKeys, 2) + for _, k := range reloadedKeys { + assert.Falsef(t, strings.HasSuffix(k, ".60000"), "stale generation-1 key should be re-keyed, got %s", k) + } + + // publish and confirm the registry and storage are fully cleaned - nothing is orphaned + testClient := setupMockClient(0) + assert.NoError(t, collector2.Execute()) + + assert.Equal(t, 2, testClient.(*MockClient).eventsAcked) + assert.Empty(t, metricRegistryGroups(collector2), "registry should be cleaned after publish") + assert.Empty(t, metricStorageKeys(collector2), "storage should be cleaned after publish (no orphans)") + + s.resetConfig() +} + +// TestMetricEventsReportedWithOwnGenerationStartTime verifies that when the registry holds metric groups +// from more than one generation, each published event is stamped with its own generation's start time +// rather than all sharing the current publish cycle's start time. +func TestMetricEventsReportedWithOwnGenerationStartTime(t *testing.T) { + defer cleanUpCachedMetricFile() + s := &testHTTPServer{} + defer s.closeServer() + s.startServer() + traceability.SetDataDirPath(".") + + metricCollector, _ := setupMetricCollectorTest(t, s) + traceStatus = healthcheck.OK + runTestHealthcheck() + + metricCollector.registry = newRegistry() + freshStorage := newStorageCache(metricCollector).(*cacheStorage) + freshStorage.isInitialized = true + metricCollector.storage = freshStorage + + addSuccess := func() { + metricCollector.AddMetricDetail(Detail{ + APIDetails: apiDetails1, + StatusCode: "200", + Duration: 10, + Bytes: 10, + AppDetails: models.AppDetails{ID: "app-1", Name: testManagedApp1}, + }) + } + + // two generations for the same app/api/status coexist in the registry (e.g. an earlier generation + // that was not yet acked), each under its own start time + metricCollector.metricStartTime = time.UnixMilli(60000) + addSuccess() + metricCollector.metricStartTime = time.UnixMilli(120000) + addSuccess() + + // publish; the current start time (120000) is >= both generations, so both are reported this cycle + testClient := setupMockClient(0) + assert.NoError(t, metricCollector.Execute()) + + mock := testClient.(*MockClient) + assert.Equal(t, 2, mock.eventsAcked) + + // each event carries its own generation's start time, not a single shared publish-cycle start time + starts := map[int64]bool{} + for _, event := range mock.capturedEvents { + raw, ok := event.Content.Fields[messageKey].(string) + if !ok { + continue + } + var v4 map[string]any + if err := json.Unmarshal([]byte(raw), &v4); err != nil { + continue + } + if ts, ok := v4["timestamp"].(float64); ok { + starts[int64(ts)] = true + } + } + assert.Equal(t, map[int64]bool{60000: true, 120000: true}, starts) + + s.resetConfig() +} + func setupAPIMetricCollectorTest(t *testing.T, s *testHTTPServer) *collector { t.Helper() cfg := createCentralCfg(s.server.URL, "demo")