fix(sinker): add activity in metric push in sinker, on change from id…

…le/unknown to active (#2221)

* fix(sinker): fix active sink not updating

Signed-off-by: Luiz Pegoraro <luiz.pegoraro@encora.com>

* Revert "fix(sinker): fix active sink not updating"

This reverts commit 5470945.

* fix(maestro): alternative fix on sink creation.

Signed-off-by: Luiz Pegoraro <luiz.pegoraro@encora.com>

* fix(maestro): updating timestamp.

Signed-off-by: Luiz Pegoraro <luiz.pegoraro@encora.com>

---------

Signed-off-by: Luiz Pegoraro <luiz.pegoraro@encora.com>
Co-authored-by: Everton H. Taques <97463920+etaques@users.noreply.github.com>

sinker/otel/bridgeservice/bridge.go

@@ -42,12 +42,20 @@ func (bs *SinkerOtelBridgeService) NotifyActiveSink(ctx context.Context, mfOwner
 		bs.logger.Error("unable to retrieve the sink config", zap.Error(err))
 		return err
 	}
-	if cfgRepo.State == config.Idle || cfgRepo.State == config.Unknown || cfgRepo.State == config.Active {
+
+	// only updates sink state if status Idle or Unknown
+	if cfgRepo.State == config.Idle || cfgRepo.State == config.Unknown {
+		cfgRepo.LastRemoteWrite = time.Now()
 		err = cfgRepo.State.SetFromString(newState)
 		if err != nil {
 			bs.logger.Error("unable to set state", zap.String("new_state", newState), zap.Error(err))
 			return err
 		}
+		err = bs.sinkerCache.AddActivity(mfOwnerId, sinkId)
+		if err != nil {
+			bs.logger.Error("error during update last remote write", zap.String("sinkId", sinkId), zap.Error(err))
+			return err
+		}
 		err = bs.sinkerCache.DeployCollector(ctx, cfgRepo)
 		if err != nil {
 			bs.logger.Error("error during update sink cache", zap.String("sinkId", sinkId), zap.Error(err))