[metric] #3 Reset declared_resource to 0 when commit updates

The current OpenCensus library keeps a metric stream that is uniquely identified by commit alive even though no new values are issued.

This makes the Otel Collector metricstransform processor receives all time value when calculating the max and gives false output.

The temporary work around is to reset the stream of previous commit to 0 when parser sees new commit.

b/321875474

e2e/testcases/otel_collector_test.go

@@ -258,7 +258,7 @@ func TestGCMMetrics(t *testing.T) {
 	if err != nil {
 		nt.T.Fatal(err)
 	}
-	_, err = retry.Retry(60*time.Second, func() error {
+	_, err = retry.Retry(120*time.Second, func() error {
 		var err error
 		for _, metricType := range GCMMetricTypes {
 			descriptor := fmt.Sprintf("%s/%s", GCMMetricPrefix, metricType)
@@ -270,6 +270,54 @@ func TestGCMMetrics(t *testing.T) {
 	if err != nil {
 		nt.T.Fatal(err)
 	}
+
+	nt.T.Log("Adding test namespace")
+	namespace := fake.NamespaceObject("foo")
+	nt.Must(nt.RootRepos[configsync.RootSyncName].Add("acme/ns.yaml", namespace))
+	nt.Must(nt.RootRepos[configsync.RootSyncName].CommitAndPush("Adding foo namespace"))
+	if err := nt.WatchForAllSyncs(); err != nil {
+		nt.T.Fatal(err)
+	}
+
+	nt.T.Log("Checking resource related metrics after adding test resource")
+	resourceMetrics := []string{
+		csmetrics.DeclaredResourcesName,
+		rgmetrics.ResourceCountName,
+		rgmetrics.ReadyResourceCountName,
+	}
+	_, err = retry.Retry(120*time.Second, func() error {
+		var err error
+		for _, metricType := range resourceMetrics {
+			descriptor := fmt.Sprintf("%s/%s", GCMMetricPrefix, metricType)
+			it := listMetricInGCM(ctx, nt, client, startTime, descriptor)
+			err = multierr.Append(err, validateMetricInGCM(nt, it, descriptor, nt.ClusterName, metricHasValue(3)))
+		}
+		return err
+	})
+	if err != nil {
+		nt.T.Fatal(err)
+	}
+
+	nt.T.Log("Remove the test resource")
+	nt.Must(nt.RootRepos[configsync.RootSyncName].Remove("acme/ns.yaml"))
+	nt.Must(nt.RootRepos[configsync.RootSyncName].CommitAndPush("Remove the test namespace"))
+	if err := nt.WatchForAllSyncs(); err != nil {
+		nt.T.Fatal(err)
+	}
+
+	nt.T.Log("Checking resource related metrics after removing test resource")
+	_, err = retry.Retry(120*time.Second, func() error {
+		var err error
+		for _, metricType := range resourceMetrics {
+			descriptor := fmt.Sprintf("%s/%s", GCMMetricPrefix, metricType)
+			it := listMetricInGCM(ctx, nt, client, startTime, descriptor)
+			err = multierr.Append(err, validateMetricInGCM(nt, it, descriptor, nt.ClusterName, metricHasLastestValue(2)))
+		}
+		return err
+	})
+	if err != nil {
+		nt.T.Fatal(err)
+	}
 }
 
 // TestOtelCollectorGCMLabelAggregation validates that Google Cloud Monitoring
@@ -451,6 +499,29 @@ func metricDoesNotHaveLabel(label string) metricValidatorFunc {
 	}
 }
 
+func metricHasValue(value int64) metricValidatorFunc {
+	return func(series *monitoringpb.TimeSeries) error {
+		points := series.GetPoints()
+		for _, point := range points {
+			if point.GetValue().GetInt64Value() == value {
+				return nil
+			}
+		}
+		return fmt.Errorf("expected metric to have value %v but did not find in response", value)
+	}
+}
+
+func metricHasLastestValue(value int64) metricValidatorFunc {
+	return func(series *monitoringpb.TimeSeries) error {
+		points := series.GetPoints()
+		lastPoint := points[len(points)-1]
+		if lastPoint.GetValue().GetInt64Value() == value {
+			return nil
+		}
+		return fmt.Errorf("expected metric to have latest value %v but did not find in response", value)
+	}
+}
+
 // Validates a metricType from a specific cluster_name can be found within given
 // TimeSeries
 func validateMetricInGCM(nt *nomostest.NT, it *monitoringv2.TimeSeriesIterator, metricType, clusterName string, valFns ...metricValidatorFunc) error {

pkg/declared/resources.go

@@ -42,6 +42,8 @@ type Resources struct {
 	objectMap *orderedmap.OrderedMap[core.ID, *unstructured.Unstructured]
 	// commit of the source in which the resources were declared
 	commit string
+	// previousCommit is the preceding commit to the commit
+	previousCommit string
 }
 
 // Update performs an atomic update on the resource declaration set.
@@ -68,95 +70,85 @@ func (r *Resources) Update(ctx context.Context, objects []client.Object, commit
 
 	// Record the declared_resources metric, after parsing but before validation.
 	metrics.RecordDeclaredResources(ctx, commit, len(newObjects))
+	if r.previousCommit != commit && r.previousCommit != "" {
+		// For Cloud Monitoring, we have configured otel-collector to remove the
+		// commit label, to reduce cardinality, but this aggregation uses the max
+		// value (b/321875474). So in order for the latest commit to be chosen as
+		// the max value, we reset the previous commit value to zero.
+		// TODO: Remove this workaround after migrating to the otel-collector metrics client and switching from gauge to async gauge
+		metrics.RecordDeclaredResources(ctx, r.previousCommit, 0)
+	}
+
+	r.mutex.Lock()
+	previousSet := r.objectMap
+	r.previousCommit = commit
+	r.objectMap = newSet
+	r.commit = commit
+	r.mutex.Unlock()
 
-	previousSet, _ := r.getObjectMap()
 	if err := deletesAllNamespaces(previousSet, newSet); err != nil {
 		return nil, err
 	}
 
-	// Now assign the pointer for the new map to the struct reference in a
-	// threadsafe context. From now on, this map is read-only.
-	r.setObjectMap(newSet, commit)
 	return newObjects, nil
 }
 
 // Get returns a copy of the resource declaration as read from Git
 func (r *Resources) Get(id core.ID) (*unstructured.Unstructured, string, bool) {
-	objSet, commit := r.getObjectMap()
-	if objSet == nil || objSet.Len() == 0 {
-		return nil, commit, false
+	r.mutex.RLock()
+	defer r.mutex.RUnlock()
+	if r.objectMap == nil || r.objectMap.Len() == 0 {
+		return nil, r.commit, false
 	}
-
-	// A local reference to the map is threadsafe since only the struct reference
-	// is replaced on update.
-	u, found := objSet.Get(id)
+	u, found := r.objectMap.Get(id)
 	// We return a copy of the Unstructured, as
 	// 1) client.Client methods mutate the objects passed into them.
 	// 2) We don't want to persist any changes made to an object we retrieved
 	//  from a declared.Resources.
-	return u.DeepCopy(), commit, found
+	return u.DeepCopy(), r.commit, found
 }
 
 // DeclaredUnstructureds returns all resource objects declared in the source,
 // along with the source commit.
 func (r *Resources) DeclaredUnstructureds() ([]*unstructured.Unstructured, string) {
-	objSet, commit := r.getObjectMap()
-	if objSet == nil || objSet.Len() == 0 {
-		return nil, commit
+	r.mutex.RLock()
+	defer r.mutex.RUnlock()
+	if r.objectMap == nil || r.objectMap.Len() == 0 {
+		return nil, r.commit
 	}
-
-	// A local reference to the map is threadsafe since only the struct reference
-	// is replaced on update.
 	var objects []*unstructured.Unstructured
-	for pair := objSet.Front(); pair != nil; pair = pair.Next() {
+	for pair := r.objectMap.Front(); pair != nil; pair = pair.Next() {
 		objects = append(objects, pair.Value)
 	}
-	return objects, commit
+	return objects, r.commit
 }
 
 // DeclaredObjects returns all resource objects declared in the source, along
 // with the source commit.
 func (r *Resources) DeclaredObjects() ([]client.Object, string) {
-	objSet, commit := r.getObjectMap()
-	if objSet == nil || objSet.Len() == 0 {
-		return nil, commit
+	r.mutex.RLock()
+	defer r.mutex.RUnlock()
+	if r.objectMap == nil || r.objectMap.Len() == 0 {
+		return nil, r.commit
 	}
-
-	// A local reference to the map is threadsafe since only the struct reference
-	// is replaced on update.
 	var objects []client.Object
-	for pair := objSet.Front(); pair != nil; pair = pair.Next() {
+	for pair := r.objectMap.Front(); pair != nil; pair = pair.Next() {
 		objects = append(objects, pair.Value)
 	}
-	return objects, commit
+	return objects, r.commit
 }
 
 // DeclaredGVKs returns the set of all GroupVersionKind found in the source,
 // along with the source commit.
 func (r *Resources) DeclaredGVKs() (map[schema.GroupVersionKind]struct{}, string) {
-	objSet, commit := r.getObjectMap()
-	if objSet == nil || objSet.Len() == 0 {
-		return nil, commit
+	r.mutex.RLock()
+	defer r.mutex.RUnlock()
+	if r.objectMap == nil || r.objectMap.Len() == 0 {
+		return nil, r.commit
 	}
-
-	// A local reference to the objSet map is threadsafe since only the pointer to
-	// the map is replaced on update.
 	gvkSet := make(map[schema.GroupVersionKind]struct{})
-	for pair := objSet.Front(); pair != nil; pair = pair.Next() {
+	for pair := r.objectMap.Front(); pair != nil; pair = pair.Next() {
 		gvkSet[pair.Value.GroupVersionKind()] = struct{}{}
 	}
-	return gvkSet, commit
-}
-
-func (r *Resources) getObjectMap() (*orderedmap.OrderedMap[core.ID, *unstructured.Unstructured], string) {
-	r.mutex.RLock()
-	defer r.mutex.RUnlock()
-	return r.objectMap, r.commit
-}
-
-func (r *Resources) setObjectMap(objectMap *orderedmap.OrderedMap[core.ID, *unstructured.Unstructured], commit string) {
-	r.mutex.Lock()
-	defer r.mutex.Unlock()
-	r.objectMap = objectMap
-	r.commit = commit
+	return gvkSet, r.commit
 }