Fix erroneous channels close (#4733)

* Fix erroneous channels close

while we did fix the goroutine leak at bulk import, in case of errors,
unfortunately we broke the non-error path, since the silence channel
needed to be closed, so that addSilenceWorker would terminate the loop,
and and wg.Wait would work

solve this by having just one cleanup function, that gets called on
defer, but also manually before returning, ensuring the error count is
correct, and all workers have indeed been collected.

Fixes issue introduced in #4556

Signed-off-by: Guido Trotter <[email protected]>

* Add import silence cli tests

Signed-off-by: Guido Trotter <[email protected]>

---------

Signed-off-by: Guido Trotter <[email protected]>
Co-authored-by: Guido Trotter <[email protected]>

Author: ultrotter

cli/silence_import.go

@@ -98,12 +98,21 @@ func (c *silenceImportCmd) bulkImport(ctx context.Context, _ *kingpin.ParseConte
 
 	amclient := NewAlertmanagerClient(alertmanagerURL)
 	silencec := make(chan *models.PostableSilence, 100)
-	defer close(silencec)
-
 	errc := make(chan error, 100)
-	defer close(errc)
 
 	var wg sync.WaitGroup
+	cleanupDone := false
+
+	closeChannels := func() {
+		if cleanupDone {
+			return
+		}
+		close(silencec)
+		wg.Wait()
+		close(errc)
+		cleanupDone = true
+	}
+	defer closeChannels()
 	for w := 0; w < c.workers; w++ {
 		wg.Add(1)
 		go func() {
@@ -138,7 +147,7 @@ func (c *silenceImportCmd) bulkImport(ctx context.Context, _ *kingpin.ParseConte
 		count++
 	}
 
-	wg.Wait()
+	closeChannels()
 
 	if errCount > 0 {
 		return fmt.Errorf("couldn't import %v out of %v silences", errCount, count)

test/cli/acceptance.go

@@ -590,6 +590,18 @@ func (am *Alertmanager) QuerySilence(match ...string) ([]TestSilence, error) {
 	return parseSilenceQueryResponse(out)
 }
 
+// QueryExpiredSilence queries expired silences using the 'amtool silence query --expired --within' command.
+func (am *Alertmanager) QueryExpiredSilence(match ...string) ([]TestSilence, error) {
+	amURLFlag := "--alertmanager.url=" + am.getURL("/")
+	args := append([]string{amURLFlag, "silence", "query", "--expired", "--within=1h"}, match...)
+	cmd := exec.Command(amtool, args...)
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		am.t.Error("Silence query command failed: ", err)
+	}
+	return parseSilenceQueryResponse(out)
+}
+
 var silenceHeaderFields = []string{"ID", "Matchers", "Ends At", "Created By", "Comment"}
 
 func parseSilenceQueryResponse(data []byte) ([]TestSilence, error) {
@@ -658,6 +670,30 @@ func (am *Alertmanager) expireSilenceCommand(sil *TestSilence) ([]byte, error) {
 	return cmd.CombinedOutput()
 }
 
+// ExportSilences exports all silences to JSON format using 'amtool silence query -o json'.
+func (am *Alertmanager) ExportSilences() ([]byte, error) {
+	amURLFlag := "--alertmanager.url=" + am.getURL("/")
+	args := []string{amURLFlag, "silence", "query", "-o", "json"}
+	cmd := exec.Command(amtool, args...)
+	return cmd.Output()
+}
+
+// ImportSilences imports silences from a JSON file using 'amtool silence import'.
+func (am *Alertmanager) ImportSilences(filename string) ([]byte, error) {
+	amURLFlag := "--alertmanager.url=" + am.getURL("/")
+	args := []string{amURLFlag, "silence", "import", filename}
+	cmd := exec.Command(amtool, args...)
+	return cmd.CombinedOutput()
+}
+
+// ExpireSilenceByID expires a silence by its ID using 'amtool silence expire'.
+func (am *Alertmanager) ExpireSilenceByID(id string) ([]byte, error) {
+	amURLFlag := "--alertmanager.url=" + am.getURL("/")
+	args := []string{amURLFlag, "silence", "expire", id}
+	cmd := exec.Command(amtool, args...)
+	return cmd.CombinedOutput()
+}
+
 // UpdateConfig rewrites the configuration file for the Alertmanager cluster. It
 // does not initiate config reloading.
 func (amc *AlertmanagerCluster) UpdateConfig(conf string) {

test/cli/acceptance/cli_test.go

@@ -271,3 +271,129 @@ receivers:
 	require.EqualError(t, err, "exit status 1")
 	require.Equal(t, "amtool: error: Failed to parse labels: unexpected open or close brace: {foo=bar}\n\n", string(out))
 }
+
+func TestSilenceImport(t *testing.T) {
+	t.Parallel()
+
+	conf := `
+route:
+  receiver: "default"
+  group_by: [alertname]
+  group_wait:      1s
+  group_interval:  1s
+  repeat_interval: 1ms
+
+receivers:
+- name: "default"
+  webhook_configs:
+  - url: 'http://%s'
+    send_resolved: true
+`
+
+	at := NewAcceptanceTest(t, &AcceptanceOpts{
+		Tolerance: 1 * time.Second,
+	})
+	co := at.Collector("webhook")
+	wh := NewWebhook(co)
+
+	amc := at.AlertmanagerCluster(fmt.Sprintf(conf, wh.Address()), 1)
+	require.NoError(t, amc.Start())
+	defer amc.Terminate()
+
+	am := amc.Members()[0]
+
+	// Add some test silences
+	silence1 := Silence(0, 4).Match("alertname=test1", "severity=warning").Comment("test silence 1")
+	silence2 := Silence(0, 4).Match("alertname=test2", "severity=critical").Comment("test silence 2")
+
+	am.SetSilence(0, silence1)
+	am.SetSilence(0, silence2)
+
+	// Export silences to JSON file
+	tmpDir := t.TempDir()
+	exportFile := tmpDir + "/silences.json"
+
+	exportOut, err := am.ExportSilences()
+	require.NoError(t, err)
+
+	// Write to file
+	err = os.WriteFile(exportFile, exportOut, 0o644)
+	require.NoError(t, err)
+
+	// Query current silences to get their IDs, then expire them
+	sils, err := am.QuerySilence()
+	require.NoError(t, err)
+	require.Len(t, sils, 2)
+	silIDs := make([]string, 0, len(sils))
+
+	// Expire all silences by ID
+	for _, sil := range sils {
+		id := sil.ID()
+		_, err := am.ExpireSilenceByID(id)
+		require.NoError(t, err)
+		silIDs = append(silIDs, id)
+	}
+
+	// Verify silences show as expired
+	sils, err = am.QueryExpiredSilence()
+	require.NoError(t, err)
+	// Silences should still be queryable but in expired state
+	require.Len(t, sils, 2, "expired silences should still be queryable")
+	// Check that the silences are actually expired (endsAt is in the past or equal to now)
+	now := float64(time.Now().Unix())
+	for _, sil := range sils {
+		require.Contains(t, silIDs, sil.ID(), "silence ID should be in the expired list")
+		require.LessOrEqual(t, sil.EndsAt(), now, "silence %s should be expired", sil.ID())
+	}
+
+	// Import silences back
+	importOut, err := am.ImportSilences(exportFile)
+	require.NoError(t, err, "import failed: %s", string(importOut))
+
+	// Verify silences were imported
+	sils, err = am.QuerySilence()
+	require.NoError(t, err)
+	require.GreaterOrEqual(t, len(sils), 2, "expected at least 2 silences after import")
+}
+
+func TestSilenceImportInvalidJSON(t *testing.T) {
+	t.Parallel()
+
+	conf := `
+route:
+  receiver: "default"
+  group_by: [alertname]
+  group_wait:      1s
+  group_interval:  1s
+  repeat_interval: 1ms
+
+receivers:
+- name: "default"
+  webhook_configs:
+  - url: 'http://%s'
+    send_resolved: true
+`
+
+	at := NewAcceptanceTest(t, &AcceptanceOpts{
+		Tolerance: 1 * time.Second,
+	})
+	co := at.Collector("webhook")
+	wh := NewWebhook(co)
+
+	amc := at.AlertmanagerCluster(fmt.Sprintf(conf, wh.Address()), 1)
+	require.NoError(t, amc.Start())
+	defer amc.Terminate()
+
+	am := amc.Members()[0]
+
+	// Create file with invalid JSON
+	tmpDir := t.TempDir()
+	invalidFile := tmpDir + "/invalid.json"
+	err := os.WriteFile(invalidFile, []byte(`[{"broken": "json"`), 0o644)
+	require.NoError(t, err)
+
+	// Try to import - should fail
+	out, err := am.ImportSilences(invalidFile)
+	require.Error(t, err, "import should fail with invalid JSON")
+	require.Contains(t, string(out), "couldn't unmarshal", "error message should mention JSON parsing")
+}

test/cli/mock.go

@@ -111,6 +111,11 @@ func (s *TestSilence) ID() string {
 	return s.id
 }
 
+// EndsAt gets the silence end time.
+func (s *TestSilence) EndsAt() float64 {
+	return s.endsAt
+}
+
 // TestAlert models a model.Alert with relative times.
 type TestAlert struct {
 	labels           models.LabelSet