[amtool] - Add a new silence import command (#1082)

* Add a new `silence import` command to amtool

This command read silences data from a query JSON output and import to
alertmanager. It allows `amtool` to be used as a backup/restore tool for
silences, i.e. #1000

Backup / export:

```
amtool silence -o json > silences.json
```

Restore / import:

```
amtool silence import silences.json
```

* Add a WaitGroup barrier

Move error channel reading to a goroutine to prevent deadlock and thus
add a WaitGroup to synchronize.

cli/silence.go

@@ -31,4 +31,5 @@ func init() {
 	silenceCmd.AddCommand(addCmd)
 	silenceCmd.AddCommand(expireCmd)
 	silenceCmd.AddCommand(queryCmd)
+	silenceCmd.AddCommand(importCmd)
 }

cli/silence_add.go

@@ -134,34 +134,43 @@ func add(cmd *cobra.Command, args []string) error {
 		Comment:   comment,
 	}
 
-	u, err := GetAlertmanagerURL()
+	silenceId, err := addSilence(&silence)
 	if err != nil {
 		return err
 	}
+
+	_, err = fmt.Println(silenceId)
+	return err
+}
+
+func addSilence(silence *types.Silence) (string, error) {
+	u, err := GetAlertmanagerURL()
+	if err != nil {
+		return "", err
+	}
 	u.Path = path.Join(u.Path, "/api/v1/silences")
 
 	buf := bytes.NewBuffer([]byte{})
 	err = json.NewEncoder(buf).Encode(silence)
 	if err != nil {
-		return err
+		return "", err
 	}
 
 	res, err := http.Post(u.String(), "application/json", buf)
 	if err != nil {
-		return err
+		return "", err
 	}
 
 	defer res.Body.Close()
 	response := addResponse{}
 	err = json.NewDecoder(res.Body).Decode(&response)
 	if err != nil {
-		return fmt.Errorf("unable to parse silence json response from %s", u.String())
+		return "", fmt.Errorf("unable to parse silence json response from %s", u.String())
 	}
 
 	if response.Status == "error" {
-		fmt.Printf("[%s] %s\n", response.ErrorType, response.Error)
-	} else {
-		fmt.Println(response.Data.SilenceID)
+		return "", fmt.Errorf("[%s] %s", response.ErrorType, response.Error)
 	}
-	return nil
+
+	return response.Data.SilenceID, nil
 }

cli/silence_import.go

@@ -0,0 +1,130 @@
+package cli
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+
+	"github.com/pkg/errors"
+	"github.com/prometheus/alertmanager/types"
+	"github.com/spf13/cobra"
+	flag "github.com/spf13/pflag"
+	"sync"
+)
+
+var importFlags *flag.FlagSet
+var importCmd = &cobra.Command{
+	Use:   "import [JSON file]",
+	Short: "Import silences",
+	Long: `Import alertmanager silences from JSON file or stdin
+
+  This command can be used to bulk import silences from a JSON file
+  created by query command. For example:
+
+  amtool silence query -o json foo > foo.json
+  amtool silence import foo.json
+
+  JSON data can also come from stdin if no param is specified.
+	`,
+	Args: cobra.MaximumNArgs(1),
+	Run:  CommandWrapper(bulkImport),
+}
+
+func init() {
+	importCmd.Flags().BoolP("force", "f", false, "Force adding new silences even if it already exists")
+	importCmd.Flags().IntP("worker", "w", 8, "Number of concurrent workers to use for import")
+	importFlags = importCmd.Flags()
+}
+
+func addSilenceWorker(silencec <-chan *types.Silence, errc chan<- error) {
+	for s := range silencec {
+		silenceId, err := addSilence(s)
+		sid := s.ID
+		if err != nil && err.Error() == "[bad_data] not found" {
+			// silence doesn't exists yet, retry to create as a new one
+			s.ID = ""
+			silenceId, err = addSilence(s)
+		}
+
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "Error adding silence id='%v': %v", sid, err)
+		} else {
+			fmt.Println(silenceId)
+		}
+		errc <- err
+	}
+}
+
+func bulkImport(cmd *cobra.Command, args []string) error {
+	force, err := importFlags.GetBool("force")
+	if err != nil {
+		return err
+	}
+
+	workers, err := importFlags.GetInt("worker")
+	if err != nil {
+		return err
+	}
+
+	input := os.Stdin
+	if len(args) == 1 {
+		input, err = os.Open(args[0])
+		if err != nil {
+			return err
+		}
+		defer input.Close()
+	}
+
+	dec := json.NewDecoder(input)
+	// read open square bracket
+	_, err = dec.Token()
+	if err != nil {
+		return errors.Wrap(err, "couldn't unmarshal input data, is it JSON?")
+	}
+
+	silencec := make(chan *types.Silence, 100)
+	errc := make(chan error, 100)
+	var wg sync.WaitGroup
+	for w := 0; w < workers; w++ {
+		go func() {
+			wg.Add(1)
+			addSilenceWorker(silencec, errc)
+			wg.Done()
+		}()
+	}
+
+	errCount := 0
+	go func() {
+		for err := range errc {
+			if err != nil {
+				errCount++
+			}
+		}
+	}()
+
+	count := 0
+	for dec.More() {
+		var s types.Silence
+		err := dec.Decode(&s)
+		if err != nil {
+			return errors.Wrap(err, "couldn't unmarshal input data, is it JSON?")
+		}
+
+		if force {
+			// reset the silence ID so Alertmanager will always create new silence
+			s.ID = ""
+		}
+
+		silencec <- &s
+		count++
+	}
+
+	close(silencec)
+	wg.Wait()
+	close(errc)
+
+	if errCount > 0 {
+		return fmt.Errorf("couldn't import %v out of %v silences", errCount, count)
+	}
+	return nil
+}