APIGOV-29006 - metric event refactor and custom unit reporting (#847)

* add new struct for new metric event

* updates to processing for handling new metric structure

* update cache to lookup managed app by catalog app id

* split metric key func

* custom registry for histo, count, grouped metrics

* update cache load for custom units

* remove start time from cached metric

* updates to group api/app/sub metrics to single grouped metric

* concurrent map write handling

* update getting proper quota ref for unit

* lookup api service id in cache

* api service id lookup

* remove TODO from testing

pkg/agent/cache/managedapplication.go

@@ -2,6 +2,8 @@ package cache
 
 import (
 	v1 "github.com/Axway/agent-sdk/pkg/apic/apiserver/models/api/v1"
+	catalog "github.com/Axway/agent-sdk/pkg/apic/apiserver/models/catalog/v1alpha1"
+	management "github.com/Axway/agent-sdk/pkg/apic/apiserver/models/management/v1alpha1"
 )
 
 // ManagedApplication cache related methods
@@ -16,7 +18,14 @@ func (c *cacheManager) AddManagedApplication(resource *v1.ResourceInstance) {
 	if resource == nil {
 		return
 	}
+	manApp := management.ManagedApplication{}
+	err := manApp.FromInstance(resource)
+	if err != nil {
+		return
+	}
+	catalogAppRef := manApp.GetReferenceByGVK(catalog.ApplicationGVK())
 	c.managedApplicationMap.SetWithSecondaryKey(resource.Metadata.ID, resource.Name, resource)
+	c.managedApplicationMap.SetSecondaryKey(resource.Metadata.ID, catalogAppRef.ID)
 }
 
 func (c *cacheManager) GetManagedApplication(id string) *v1.ResourceInstance {
@@ -33,6 +42,20 @@ func (c *cacheManager) GetManagedApplication(id string) *v1.ResourceInstance {
 	return nil
 }
 
+func (c *cacheManager) GetManagedApplicationByApplicationID(id string) *v1.ResourceInstance {
+	c.ApplyResourceReadLock()
+	defer c.ReleaseResourceReadLock()
+
+	managedApp, _ := c.managedApplicationMap.GetBySecondaryKey(id)
+	if managedApp != nil {
+		ri, ok := managedApp.(*v1.ResourceInstance)
+		if ok {
+			return ri
+		}
+	}
+	return nil
+}
+
 func (c *cacheManager) GetManagedApplicationByName(name string) *v1.ResourceInstance {
 	c.ApplyResourceReadLock()
 	defer c.ReleaseResourceReadLock()

pkg/agent/cache/manager.go

@@ -92,6 +92,7 @@ type Manager interface {
 	GetManagedApplicationCacheKeys() []string
 	AddManagedApplication(resource *v1.ResourceInstance)
 	GetManagedApplication(id string) *v1.ResourceInstance
+	GetManagedApplicationByApplicationID(id string) *v1.ResourceInstance
 	GetManagedApplicationByName(name string) *v1.ResourceInstance
 	DeleteManagedApplication(id string) error
 
@@ -116,7 +117,6 @@ type Manager interface {
 	ReleaseResourceReadLock()
 }
 
-type teamRefreshHandler func()
 type cacheManager struct {
 	jobs.Job
 	logger                  log.FieldLogger
@@ -157,6 +157,7 @@ func NewAgentCacheManager(cfg config.CentralConfig, persistCacheEnabled bool) Ma
 		m.migrators = []cacheMigrate{
 			m.migrateAccessRequest,
 			m.migrateInstanceCount,
+			m.migrateManagedApplications,
 		}
 	}
 	m.initializeCache(cfg)

pkg/agent/cache/migratepersistedcache.go

@@ -4,6 +4,7 @@ import (
 	"sync"
 
 	v1 "github.com/Axway/agent-sdk/pkg/apic/apiserver/models/api/v1"
+	catalog "github.com/Axway/agent-sdk/pkg/apic/apiserver/models/catalog/v1alpha1"
 	management "github.com/Axway/agent-sdk/pkg/apic/apiserver/models/management/v1alpha1"
 	defs "github.com/Axway/agent-sdk/pkg/apic/definitions"
 	"github.com/Axway/agent-sdk/pkg/util"
@@ -105,3 +106,28 @@ func (c *cacheManager) migrateInstanceCount(key string) error {
 	}
 	return nil
 }
+
+func (c *cacheManager) migrateManagedApplications(cacheKey string) error {
+	if cacheKey != managedAppKey {
+		return nil
+	}
+
+	for _, key := range c.managedApplicationMap.GetKeys() {
+		cachedManagedApp, _ := c.managedApplicationMap.Get(key)
+		if cachedManagedApp == nil {
+			continue
+		}
+		ri, ok := cachedManagedApp.(*v1.ResourceInstance)
+		if !ok {
+			continue
+		}
+		manApp := management.ManagedApplication{}
+		err := manApp.FromInstance(ri)
+		if err != nil {
+			continue
+		}
+		catalogAppRef := manApp.GetReferenceByGVK(catalog.ApplicationGVK())
+		c.managedApplicationMap.SetSecondaryKey(key, catalogAppRef.ID)
+	}
+	return nil
+}

pkg/transaction/metric/apimetric.go

@@ -16,14 +16,14 @@ type APIMetric struct {
 	AssetResource models.AssetResource `json:"assetResource,omitempty"`
 	ProductPlan   models.ProductPlan   `json:"productPlan,omitempty"`
 	Quota         models.Quota         `json:"quota,omitempty"`
-	Unit          models.Unit          `json:"unit,omitempty"`
 	StatusCode    string               `json:"statusCode,omitempty"`
 	Status        string               `json:"status,omitempty"`
 	Count         int64                `json:"count"`
 	Response      ResponseMetrics      `json:"response,omitempty"`
 	Observation   ObservationDetails   `json:"observation"`
 	EventID       string               `json:"-"`
 	StartTime     time.Time            `json:"-"`
+	Unit          *models.Unit         `json:"unit,omitempty"`
 }
 
 // GetStartTime - Returns the start time for subscription metric

pkg/transaction/metric/cachestorage.go

@@ -13,6 +13,7 @@ import (
 	"github.com/Axway/agent-sdk/pkg/agent"
 	"github.com/Axway/agent-sdk/pkg/cache"
 	"github.com/Axway/agent-sdk/pkg/traceability"
+	"github.com/Axway/agent-sdk/pkg/transaction/models"
 	"github.com/Axway/agent-sdk/pkg/util"
 	"github.com/rcrowley/go-metrics"
 )
@@ -32,8 +33,8 @@ type storageCache interface {
 	updateUsage(usageCount int)
 	updateVolume(bytes int64)
 	updateAppUsage(usageCount int, appID string)
-	updateMetric(cachedMetric cachedMetricInterface, metric *centralMetricEvent)
-	removeMetric(metric *centralMetricEvent)
+	updateMetric(cachedMetric cachedMetricInterface, metric *centralMetric)
+	removeMetric(metric *centralMetric)
 	save()
 }
 
@@ -161,20 +162,41 @@ func (c *cacheStorage) loadMetrics(storageCache cache.Cache) {
 			var cm cachedMetric
 			json.Unmarshal(buffer, &cm)
 
-			var metric *centralMetricEvent
-			for _, duration := range cm.Values {
-				unitID := ""
-				if cm.Unit != nil {
-					unitID = cm.Unit.ID
+			apiDetails := models.APIDetails{}
+			if cm.API != nil {
+				apiDetails.ID = cm.API.ID
+				apiDetails.Name = cm.API.Name
+			}
+			appDetails := models.AppDetails{}
+			if cm.App != nil {
+				appDetails.ID = cm.App.ID
+				appDetails.ConsumerOrgID = cm.App.ConsumerOrgID
+			}
+
+			if len(cm.Values) == 0 {
+				if cm.Unit == nil {
+					continue
 				}
-				metricDetail := Detail{
-					APIDetails: *cm.API,
-					AppDetails: *cm.App,
-					UnitName:   unitID,
+
+				c.collector.AddCustomMetricDetail(CustomMetricDetail{
+					APIDetails: apiDetails,
+					AppDetails: appDetails,
+					UnitDetails: models.Unit{
+						Name: cm.Unit.Name,
+					},
+					Count: cm.Count,
+				})
+				continue
+			}
+
+			var metric *centralMetric
+			for _, duration := range cm.Values {
+				metric = c.collector.createOrUpdateHistogram(Detail{
+					APIDetails: apiDetails,
+					AppDetails: appDetails,
 					StatusCode: cm.StatusCode,
 					Duration:   duration,
-				}
-				metric = c.collector.createOrUpdateMetric(metricDetail)
+				})
 			}
 
 			newKey := metric.getKey()
@@ -184,25 +206,22 @@ func (c *cacheStorage) loadMetrics(storageCache cache.Cache) {
 				c.storageLock.Unlock()
 			}
 			storageCache.Set(newKey, cm)
-			if metric != nil {
-				metric.StartTime = cm.StartTime
-			}
 		}
 	}
 }
 
-func (c *cacheStorage) updateMetric(cached cachedMetricInterface, metric *centralMetricEvent) {
+func (c *cacheStorage) updateMetric(cached cachedMetricInterface, metric *centralMetric) {
 	if !c.isInitialized {
 		return
 	}
 
 	c.storageLock.Lock()
 	defer c.storageLock.Unlock()
 
-	c.storage.Set(metric.getKey(), metric.createdCachedMetric(cached))
+	c.storage.Set(metric.getKey(), metric.createCachedMetric(cached))
 }
 
-func (c *cacheStorage) removeMetric(metric *centralMetricEvent) {
+func (c *cacheStorage) removeMetric(metric *centralMetric) {
 	if !c.isInitialized {
 		return
 	}