APIGOV-33037 - clean storage of metrics on publish

pkg/transaction/metric/cachestorage.go

@@ -172,23 +172,24 @@ 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{
 						Name: cm.Unit.Name,
 					},
 					Count: cm.Count,
 				})
+				c.rekeyLoadedMetric(storageCache, metric, cacheKey, cm)
 				continue
 			}
 
 			if cm.Count == 0 && len(cm.Values) == 0 {
 				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() {

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 {

pkg/transaction/metric/metricscollector.go

@@ -305,24 +305,28 @@ 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).
 		WithField("unitName", detail.UnitDetails.Name)
 
 	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()
 }