Handle action requests for collectors (#444)

An action is retrieved per collector ID.
We need to iterate over all backends for a collector and execute that
action.

Author: mpfz0r

backends/backend.go

@@ -36,6 +36,7 @@ type Backend struct {
 	Enabled              *bool
 	Id                   string
 	ConfigId             string
+	CollectorId          string
 	Name                 string
 	ServiceType          string
 	OperatingSystem      string
@@ -51,6 +52,7 @@ func BackendFromResponse(response graylog.ResponseCollectorBackend, configId str
 	return &Backend{
 		Enabled:              common.NewTrue(),
 		Id:                   response.Id + "-" + configId,
+		CollectorId:          response.Id,
 		ConfigId:             configId,
 		Name:                 response.Name + "-" + configId,
 		ServiceType:          response.ServiceType,
@@ -83,6 +85,7 @@ func (b *Backend) EqualSettings(a *Backend) bool {
 		Enabled:              b.Enabled,
 		Id:                   a.Id,
 		ConfigId:             a.ConfigId,
+		CollectorId:          a.CollectorId,
 		Name:                 a.Name,
 		ServiceType:          a.ServiceType,
 		OperatingSystem:      a.OperatingSystem,

backends/registry.go

@@ -46,6 +46,16 @@ func (bs *backendStore) SetBackend(backend Backend) {
 	bs.backends[backend.Id].ValidationParameters = validationParameters
 }
 
+func (bs *backendStore) GetBackendsForCollectorId(collectorId string) []*Backend {
+	var backends []*Backend
+	for _, backend := range bs.backends {
+		if backend.CollectorId == collectorId {
+			backends = append(backends, backend)
+		}
+	}
+	return backends
+}
+
 func (bs *backendStore) GetBackend(id string) *Backend {
 	return bs.backends[id]
 }

daemon/action_handler.go

@@ -23,21 +23,22 @@ import (
 
 func HandleCollectorActions(actions []graylog.ResponseCollectorAction) {
 	for _, action := range actions {
-		backend := backends.Store.GetBackend(action.BackendId)
-		if backend == nil {
+		backends := backends.Store.GetBackendsForCollectorId(action.BackendId)
+		if backends == nil {
 			log.Errorf("Got action for non-existing collector: %s", action.BackendId)
 			continue
 		}
-
-		switch {
-		case action.Properties["start"] == true:
-			startAction(backend)
-		case action.Properties["restart"] == true:
-			restartAction(backend)
-		case action.Properties["stop"] == true:
-			stopAction(backend)
-		default:
-			log.Infof("Got unsupported collector command: %s", common.Inspect(action.Properties))
+		for _, backend := range backends {
+			switch {
+			case action.Properties["start"] == true:
+				startAction(backend)
+			case action.Properties["restart"] == true:
+				restartAction(backend)
+			case action.Properties["stop"] == true:
+				stopAction(backend)
+			default:
+				log.Infof("Got unsupported collector command: %s", common.Inspect(action.Properties))
+			}
 		}
 	}
 }