fix(sumologicextension): End the heartbeatloop immediately on parent …

…shutdown using context

.chloggen/fix-sumo-extension.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: 'bug_fix'
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: sumologicextension
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Fixes an issue where there was a delay in stopping heartbeat messages when the extension was stopped."
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [32785]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

extension/sumologicextension/extension.go

@@ -204,7 +204,7 @@ func (se *SumologicExtension) Start(ctx context.Context, host component.Host) er
 		}
 	}
 
-	go se.heartbeatLoop()
+	go se.heartbeatLoop(ctx)
 
 	return nil
 }
@@ -556,17 +556,16 @@ func (se *SumologicExtension) registerCollectorWithBackoff(ctx context.Context,
 	}
 }
 
-func (se *SumologicExtension) heartbeatLoop() {
+func (se *SumologicExtension) heartbeatLoop(ctx context.Context) {
 	if se.registrationInfo.CollectorCredentialID == "" || se.registrationInfo.CollectorCredentialKey == "" {
 		se.logger.Error("Collector not registered, cannot send heartbeat")
 		return
 	}
 
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
+	ctx, cancel := context.WithCancel(ctx)
 	go func() {
-		// When the close channel is closed ...
-		<-se.closeChan
+		// When the context is done ...
+		<-ctx.Done()
 		// ... cancel the ongoing heartbeat request.
 		cancel()
 	}()

extension/sumologicextension/extension_test.go

@@ -566,69 +566,72 @@ func TestRegisterEmptyCollectorNameForceRegistration(t *testing.T) {
 
 	hostname, err := getHostname(zap.NewNop())
 	require.NoError(t, err)
-	srv := httptest.NewServer(func() http.HandlerFunc {
+	getServer := func() *httptest.Server {
 		var reqCount int32
 
-		return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
-			// TODO Add payload verification - verify if collectorName is set properly
-			reqNum := atomic.AddInt32(&reqCount, 1)
+		return httptest.NewServer(http.HandlerFunc(
 
-			switch reqNum {
+			func(w http.ResponseWriter, req *http.Request) {
+				// TODO Add payload verification - verify if collectorName is set properly
+				reqNum := atomic.AddInt32(&reqCount, 1)
 
-			// register
-			case 1:
-				require.Equal(t, registerURL, req.URL.Path)
+				switch reqNum {
 
-				authHeader := req.Header.Get("Authorization")
-				assert.Equal(t, "Bearer dummy_install_token", authHeader,
-					"collector didn't send correct Authorization header with registration request")
+				// register
+				case 1:
+					require.Equal(t, registerURL, req.URL.Path)
 
-				_, err = w.Write([]byte(`{
+					authHeader := req.Header.Get("Authorization")
+					assert.Equal(t, "Bearer dummy_install_token", authHeader,
+						"collector didn't send correct Authorization header with registration request")
+
+					_, err = w.Write([]byte(`{
 					"collectorCredentialID": "aaaaaaaaaaaaaaaaaaaa",
 					"collectorCredentialKey": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
 					"collectorId": "000000000FFFFFFF",
 					"collectorName": "hostname-test-123456123123"
 				}`))
-				if err != nil {
-					w.WriteHeader(http.StatusInternalServerError)
-				}
+					if err != nil {
+						w.WriteHeader(http.StatusInternalServerError)
+					}
 
-			// metadata
-			case 2:
-				assert.Equal(t, metadataURL, req.URL.Path)
-				w.WriteHeader(200)
+				// metadata
+				case 2:
+					assert.Equal(t, metadataURL, req.URL.Path)
+					w.WriteHeader(200)
 
-			// register again because force registration was set
-			case 3:
-				require.Equal(t, registerURL, req.URL.Path)
+				// register again because force registration was set
+				case 3:
+					require.Equal(t, registerURL, req.URL.Path)
 
-				authHeader := req.Header.Get("Authorization")
-				assert.Equal(t, "Bearer dummy_install_token", authHeader,
-					"collector didn't send correct Authorization header with registration request")
+					authHeader := req.Header.Get("Authorization")
+					assert.Equal(t, "Bearer dummy_install_token", authHeader,
+						"collector didn't send correct Authorization header with registration request")
 
-				_, err = w.Write([]byte(`{
+					_, err = w.Write([]byte(`{
 					"collectorCredentialID": "aaaaaaaaaaaaaaaaaaaa",
 					"collectorCredentialKey": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
 					"collectorId": "000000000FFFFFFF",
 					"collectorName": "hostname-test-123456123123"
 				}`))
-				if err != nil {
-					w.WriteHeader(http.StatusInternalServerError)
-				}
+					if err != nil {
+						w.WriteHeader(http.StatusInternalServerError)
+					}
 
-			// metadata
-			case 4:
-				assert.Equal(t, metadataURL, req.URL.Path)
-				w.WriteHeader(200)
+				// metadata
+				case 4:
+					assert.Equal(t, metadataURL, req.URL.Path)
+					w.WriteHeader(200)
 
-			// should not produce any more requests
-			default:
-				w.WriteHeader(http.StatusInternalServerError)
-			}
-		})
-	}())
+				// should not produce any more requests
+				default:
+					w.WriteHeader(http.StatusInternalServerError)
+				}
+			}))
+	}
 
 	dir, err := os.MkdirTemp("", "otelcol-sumo-store-credentials-test-*")
+	srv := getServer()
 	t.Cleanup(func() {
 		srv.Close()
 		os.RemoveAll(dir)
@@ -651,6 +654,10 @@ func TestRegisterEmptyCollectorNameForceRegistration(t *testing.T) {
 	colCreds, err := se.credentialsStore.Get(se.hashKey)
 	require.NoError(t, err)
 	colName := colCreds.CollectorName
+	t.Cleanup(func() {
+		srv.Close()
+	})
+	srv = getServer()
 	se, err = newSumologicExtension(cfg, zap.NewNop(), component.NewID(metadata.Type), "1.0.0")
 	require.NoError(t, err)
 	require.NoError(t, se.Start(context.Background(), componenttest.NewNopHost()))