Merge pull request #824 from openconfig/coll-target-state

reflect target last error in a consistent way

Author: karimra

pkg/api/target/subscribe.go

@@ -210,9 +210,7 @@ func (t *Target) attemptSubscription(ctx context.Context, req *gnmi.SubscribeReq
 		if isCancellationError(err) {
 			return false
 		}
-		sendError(errCh, ctx, subscriptionName,
-			fmt.Errorf("failed to create subscribe client, target='%s', retry in %s: %w",
-				t.Config.Name, t.Config.RetryTimer, err))
+		sendError(errCh, ctx, subscriptionName, err)
 		return true
 	}
 
@@ -264,8 +262,6 @@ func (t *Target) handleSTREAMMode(nctx, ctx context.Context, client gnmi.GNMI_Su
 		}
 
 		sendError(errCh, ctx, subscriptionName, err)
-		sendError(errCh, ctx, subscriptionName,
-			fmt.Errorf("retrying in %s", t.Config.RetryTimer))
 		return true
 	}
 	return false
@@ -288,8 +284,6 @@ func (t *Target) handleONCEMode(nctx, ctx context.Context, client gnmi.GNMI_Subs
 			return false
 		}
 
-		sendError(errCh, ctx, subscriptionName,
-			fmt.Errorf("retrying in %s", t.Config.RetryTimer))
 		return true
 	}
 	return false
@@ -315,8 +309,8 @@ func (t *Target) handlePOLLMode(nctx, ctx context.Context, client gnmi.GNMI_Subs
 		}
 
 		sendError(errCh, ctx, subscriptionName, err)
-		sendError(errCh, ctx, subscriptionName,
-			fmt.Errorf("retrying in %s", t.Config.RetryTimer))
+		// sendError(errCh, ctx, subscriptionName,
+		// 	fmt.Errorf("retrying in %s", t.Config.RetryTimer))
 		return true
 	}
 	return false

pkg/api/target/target.go

@@ -281,6 +281,22 @@ func (t *Target) Close() error {
 	return nil
 }
 
+// SubscribeClientStates returns current subscription states.
+// based on the SubscribeClients map.
+func (t *Target) SubscribeClientStates() map[string]bool {
+	t.m.Lock()
+	defer t.m.Unlock()
+	if len(t.Subscriptions) == 0 {
+		return nil
+	}
+	states := make(map[string]bool, len(t.Subscriptions))
+	for name := range t.Subscriptions {
+		_, ok := t.SubscribeClients[name]
+		states[name] = ok
+	}
+	return states
+}
+
 func (t *Target) ConnState() string {
 	if t.conn == nil {
 		return ""

pkg/collector/api/server/subscriptions.go

@@ -169,22 +169,19 @@ func (s *Server) handleSubscriptionsGet(w http.ResponseWriter, r *http.Request)
 		}
 		// Collect all subscriptions from targets
 		s.targetsManager.ForEach(func(mt *targets_manager.ManagedTarget) {
-			for _, sub := range mt.T.Subscriptions {
-				if subscriptionsMap[sub.Name] == nil {
-					subscriptionsMap[sub.Name] = &SubscriptionResponse{
-						Name:    sub.Name,
-						Config:  sub,
+			subStates := mt.T.SubscribeClientStates()
+			for name, active := range subStates {
+				if subscriptionsMap[name] == nil {
+					subscriptionsMap[name] = &SubscriptionResponse{
+						Name:    name,
 						Targets: make(map[string]*TargetStateInfo),
 					}
 				}
-
-				// Determine state for this subscription on this target
 				state := "stopped"
-				if _, ok := mt.T.SubscribeClients[sub.Name]; ok {
+				if active {
 					state = "running"
 				}
-
-				subscriptionsMap[sub.Name].Targets[mt.Name] = &TargetStateInfo{
+				subscriptionsMap[name].Targets[mt.Name] = &TargetStateInfo{
 					Name:  mt.Name,
 					State: state,
 				}
@@ -222,20 +219,18 @@ func (s *Server) handleSubscriptionsGet(w http.ResponseWriter, r *http.Request)
 			Targets: make(map[string]*TargetStateInfo),
 		}
 		s.targetsManager.ForEach(func(mt *targets_manager.ManagedTarget) {
-			for _, sub := range mt.T.Subscriptions {
-				if sub.Name != id {
-					continue
-				}
-
-				// Determine state for this subscription on this target
-				state := "stopped"
-				if _, ok := mt.T.SubscribeClients[sub.Name]; ok {
-					state = "running"
-				}
-				response.Targets[mt.Name] = &TargetStateInfo{
-					Name:  mt.Name,
-					State: state,
-				}
+			subStates := mt.T.SubscribeClientStates()
+			active, exists := subStates[id]
+			if !exists {
+				return
+			}
+			state := "stopped"
+			if active {
+				state = "running"
+			}
+			response.Targets[mt.Name] = &TargetStateInfo{
+				Name:  mt.Name,
+				State: state,
 			}
 		})
 		err = json.NewEncoder(w).Encode(response)