More descriptive names for event structs and files

This is to eliminate some confusion around how events are handled in
Swarm.

Signed-off-by: Nishant Totla <[email protected]>

Author: nishanttotla

api/handlers.go

@@ -853,47 +853,6 @@ func postImagesLoad(c *context, w http.ResponseWriter, r *http.Request) {
 	if errorFound {
 		sendErrorJSONMessage(wf, 1, errorMessage)
 	}
-
-}
-
-// normalizeEvent takes a cluster Event and ensures backward compatibility
-// and all the right fields filled up
-func normalizeEvent(receivedEvent *cluster.Event) ([]byte, error) {
-	// make a local copy of the event
-	e := *receivedEvent
-	// make a fresh copy of the Actor.Attributes map to prevent a race condition
-	e.Actor.Attributes = make(map[string]string)
-	for k, v := range receivedEvent.Actor.Attributes {
-		e.Actor.Attributes[k] = v
-	}
-
-	// remove this hack once 1.10 is broadly adopted
-	e.From = e.From + " node:" + e.Engine.Name
-
-	e.Actor.Attributes["node.name"] = e.Engine.Name
-	e.Actor.Attributes["node.id"] = e.Engine.ID
-	e.Actor.Attributes["node.addr"] = e.Engine.Addr
-	e.Actor.Attributes["node.ip"] = e.Engine.IP
-
-	data, err := json.Marshal(&e)
-	if err != nil {
-		return nil, err
-	}
-
-	// remove the node field once 1.10 is broadly adopted & interlock stops relying on it
-	node := fmt.Sprintf(",%q:{%q:%q,%q:%q,%q:%q,%q:%q}}",
-		"node",
-		"Name", e.Engine.Name,
-		"Id", e.Engine.ID,
-		"Addr", e.Engine.Addr,
-		"Ip", e.Engine.IP,
-	)
-
-	// insert Node field
-	data = data[:len(data)-1]
-	data = append(data, []byte(node)...)
-
-	return data, nil
 }
 
 // GET /events

api/primary.go

@@ -14,7 +14,7 @@ import (
 // Primary router context, used by handlers.
 type context struct {
 	cluster       cluster.Cluster
-	eventsHandler *cluster.EventsHandler
+	eventsHandler *cluster.APIEventHandler
 	listenerCount *uint64
 	statusHandler StatusHandler
 	debug         bool
@@ -119,11 +119,11 @@ func profilerSetup(mainRouter *mux.Router, path string) {
 func NewPrimary(cluster cluster.Cluster, tlsConfig *tls.Config, status StatusHandler, debug, enableCors bool) *mux.Router {
 	// Register the API events handler in the cluster.
 
-	// NewEventsHandler creates a new eventsHandler object. The new eventsHandler
+	// NewAPIEventHandler creates a new eventsHandler object. The new eventsHandler
 	// is initialized with no writers or channels. This is in api/events.go and
 	// uses the watch package from SwarmKit, which is based on the go-events
 	// package. See https://github.com/docker/swarm/issues/2718 for context
-	eventsHandler := cluster.NewAPIEventsHandler()
+	eventsHandler := cluster.NewAPIEventHandler()
 	listenerCount := uint64(0)
 	// need to add this queue to the cluster
 	// This just calls c.eventHandlers.RegisterEventHandler(eventsHandler) internally.

api/utils.go

@@ -388,3 +388,43 @@ func matchImageOSError(errMsg string) string {
 	}
 	return results[1]
 }
+
+// normalizeEvent takes a cluster Event and ensures backward compatibility
+// and all the right fields filled up
+func normalizeEvent(receivedEvent *cluster.Event) ([]byte, error) {
+	// make a local copy of the event
+	e := *receivedEvent
+	// make a fresh copy of the Actor.Attributes map to prevent a race condition
+	e.Actor.Attributes = make(map[string]string)
+	for k, v := range receivedEvent.Actor.Attributes {
+		e.Actor.Attributes[k] = v
+	}
+
+	// remove this hack once 1.10 is broadly adopted
+	e.From = e.From + " node:" + e.Engine.Name
+
+	e.Actor.Attributes["node.name"] = e.Engine.Name
+	e.Actor.Attributes["node.id"] = e.Engine.ID
+	e.Actor.Attributes["node.addr"] = e.Engine.Addr
+	e.Actor.Attributes["node.ip"] = e.Engine.IP
+
+	data, err := json.Marshal(&e)
+	if err != nil {
+		return nil, err
+	}
+
+	// remove the node field once 1.10 is broadly adopted & interlock stops relying on it
+	node := fmt.Sprintf(",%q:{%q:%q,%q:%q,%q:%q,%q:%q}}",
+		"node",
+		"Name", e.Engine.Name,
+		"Id", e.Engine.ID,
+		"Addr", e.Engine.Addr,
+		"Ip", e.Engine.IP,
+	)
+
+	// insert Node field
+	data = data[:len(data)-1]
+	data = append(data, []byte(node)...)
+
+	return data, nil
+}

cluster/api_events.go

@@ -0,0 +1,48 @@
+package cluster
+
+import (
+	"time"
+
+	"github.com/docker/go-events"
+	"github.com/docker/swarmkit/watch"
+)
+
+const (
+	defaultEventQueueLimit   = 10000
+	defaultEventQueueTimeout = 10 * time.Second
+)
+
+// APIEventsHandler broadcasts events to multiple client listeners.
+type APIEventHandler struct {
+	watchQueue *watch.Queue
+}
+
+// NewAPIEventHandler creates a new APIEventsHandler for a cluster.
+// The new eventsHandler is initialized with no writers or channels.
+func NewAPIEventHandler() *APIEventHandler {
+	return &APIEventHandler{
+		watchQueue: watch.NewQueue(watch.WithTimeout(defaultEventQueueTimeout), watch.WithLimit(defaultEventQueueLimit), watch.WithCloseOutChan()),
+	}
+}
+
+// Add adds the writer and a new channel for the remote address.
+func (eh *APIEventHandler) Watch() (eventq chan events.Event, cancel func()) {
+	eventq, cancel = eh.watchQueue.Watch()
+	return eventq, cancel
+}
+
+func (eh *APIEventHandler) cleanupHandler(remoteAddr string) {
+	eh.watchQueue.Close()
+}
+
+// Handle writes information about a cluster event to each remote address in the cluster that has been added to the events handler.
+// After an unsuccessful write to a remote address, the associated channel is closed and the address is removed from the events handler.
+func (eh *APIEventHandler) Handle(e *Event) error {
+	eh.watchQueue.Publish(e)
+	return nil
+}
+
+// Size returns the number of remote addresses that the events handler currently contains.
+func (eh *APIEventHandler) Size() int {
+	return 0
+}

cluster/cluster.go

@@ -81,8 +81,8 @@ type Cluster interface {
 	// UnregisterEventHandler unregisters an event handler.
 	UnregisterEventHandler(h EventHandler)
 
-	// NewAPIEventsHandler creates a new API events handler
-	NewAPIEventsHandler() *EventsHandler
+	// NewAPIEventHandler creates a new API events handler
+	NewAPIEventHandler() *APIEventHandler
 
 	// CloseWatchQueue closes the watchQueue when the manager shuts down.
 	CloseWatchQueue()

cluster/event.go cluster/event_map.go

@@ -5,36 +5,23 @@ import (
 	"sync"
 
 	log "github.com/Sirupsen/logrus"
-	"github.com/docker/docker/api/types/events"
 )
 
-// Event is exported
-type Event struct {
-	events.Message
-	Engine *Engine `json:"-"`
-}
-
-// EventHandler is exported
-type EventHandler interface {
-	Handle(*Event) error
-}
-
-// EventHandlers is a map of EventHandler
-type EventHandlers struct {
+// ClusterEventHandlers is a map of EventHandler
+type ClusterEventHandlers struct {
 	sync.RWMutex
-
 	eventHandlers map[EventHandler]struct{}
 }
 
-// NewEventHandlers returns an EventHandlers
-func NewEventHandlers() *EventHandlers {
-	return &EventHandlers{
+// NewClusterEventHandlers returns an EventHandlers
+func NewClusterEventHandlers() *ClusterEventHandlers {
+	return &ClusterEventHandlers{
 		eventHandlers: make(map[EventHandler]struct{}),
 	}
 }
 
 // Handle callbacks for the events
-func (eh *EventHandlers) Handle(e *Event) {
+func (eh *ClusterEventHandlers) Handle(e *Event) {
 	eh.RLock()
 	defer eh.RUnlock()
 
@@ -46,7 +33,7 @@ func (eh *EventHandlers) Handle(e *Event) {
 }
 
 // RegisterEventHandler registers an event handler.
-func (eh *EventHandlers) RegisterEventHandler(h EventHandler) error {
+func (eh *ClusterEventHandlers) RegisterEventHandler(h EventHandler) error {
 	eh.Lock()
 	defer eh.Unlock()
 
@@ -58,7 +45,7 @@ func (eh *EventHandlers) RegisterEventHandler(h EventHandler) error {
 }
 
 // UnregisterEventHandler unregisters a previously registered event handler.
-func (eh *EventHandlers) UnregisterEventHandler(h EventHandler) {
+func (eh *ClusterEventHandlers) UnregisterEventHandler(h EventHandler) {
 	eh.Lock()
 	defer eh.Unlock()
 

cluster/events.go

@@ -1,48 +1,20 @@
 package cluster
 
-import (
-	"time"
+import "github.com/docker/docker/api/types/events"
 
-	"github.com/docker/go-events"
-	"github.com/docker/swarmkit/watch"
-)
-
-const (
-	defaultEventQueueLimit   = 10000
-	defaultEventQueueTimeout = 10 * time.Second
-)
-
-// EventsHandler broadcasts events to multiple client listeners.
-type EventsHandler struct {
-	watchQueue *watch.Queue
+// Event is exported
+type Event struct {
+	events.Message
+	Engine *Engine `json:"-"`
 }
 
-// NewEventsHandler creates a new EventsHandler for a cluster.
-// The new eventsHandler is initialized with no writers or channels.
-func NewEventsHandler() *EventsHandler {
-	return &EventsHandler{
-		watchQueue: watch.NewQueue(watch.WithTimeout(defaultEventQueueTimeout), watch.WithLimit(defaultEventQueueLimit), watch.WithCloseOutChan()),
-	}
+// EventHandler is exported
+type EventHandler interface {
+	Handle(*Event) error
 }
 
-// Add adds the writer and a new channel for the remote address.
-func (eh *EventsHandler) Watch() (eventq chan events.Event, cancel func()) {
-	eventq, cancel = eh.watchQueue.Watch()
-	return eventq, cancel
-}
-
-func (eh *EventsHandler) cleanupHandler(remoteAddr string) {
-	eh.watchQueue.Close()
-}
-
-// Handle writes information about a cluster event to each remote address in the cluster that has been added to the events handler.
-// After an unsuccessful write to a remote address, the associated channel is closed and the address is removed from the events handler.
-func (eh *EventsHandler) Handle(e *Event) error {
-	eh.watchQueue.Publish(e)
-	return nil
-}
-
-// Size returns the number of remote addresses that the events handler currently contains.
-func (eh *EventsHandler) Size() int {
-	return 0
-}
+// EventHandler is implemented by all event handlers in Swarm
+// - APIEventHandler: Handles API level events
+// - Watchdog: Handles events related to rescheduling
+// - Cluster: Acts as a proxy event handler for the engine, but essentially
+// punts all handling to the above two handlers

cluster/mesos/cluster.go

@@ -31,18 +31,18 @@ import (
 type Cluster struct {
 	sync.RWMutex
 
-	dockerEnginePort    string
-	eventHandlers       *cluster.EventHandlers
-	master              string
-	agents              map[string]*agent
-	scheduler           *Scheduler
-	TLSConfig           *tls.Config
-	options             *cluster.DriverOpts
-	offerTimeout        time.Duration
-	refuseTimeout       time.Duration
-	taskCreationTimeout time.Duration
-	pendingTasks        *task.Tasks
-	engineOpts          *cluster.EngineOpts
+	dockerEnginePort     string
+	clusterEventHandlers *cluster.ClusterEventHandlers
+	master               string
+	agents               map[string]*agent
+	scheduler            *Scheduler
+	TLSConfig            *tls.Config
+	options              *cluster.DriverOpts
+	offerTimeout         time.Duration
+	refuseTimeout        time.Duration
+	taskCreationTimeout  time.Duration
+	pendingTasks         *task.Tasks
+	engineOpts           *cluster.EngineOpts
 }
 
 const (
@@ -69,16 +69,16 @@ func NewCluster(scheduler *scheduler.Scheduler, TLSConfig *tls.Config, master st
 		flag.Lookup("logtostderr").Value.Set("true")
 	}
 	cluster := &Cluster{
-		dockerEnginePort:    defaultDockerEnginePort,
-		eventHandlers:       cluster.NewEventHandlers(),
-		master:              master,
-		agents:              make(map[string]*agent),
-		TLSConfig:           TLSConfig,
-		options:             &options,
-		offerTimeout:        defaultOfferTimeout,
-		taskCreationTimeout: defaultTaskCreationTimeout,
-		engineOpts:          engineOptions,
-		refuseTimeout:       defaultRefuseTimeout,
+		dockerEnginePort:     defaultDockerEnginePort,
+		clusterEventHandlers: cluster.NewClusterEventHandlers(),
+		master:               master,
+		agents:               make(map[string]*agent),
+		TLSConfig:            TLSConfig,
+		options:              &options,
+		offerTimeout:         defaultOfferTimeout,
+		taskCreationTimeout:  defaultTaskCreationTimeout,
+		engineOpts:           engineOptions,
+		refuseTimeout:        defaultRefuseTimeout,
 	}
 
 	cluster.pendingTasks = task.NewTasks(cluster)
@@ -167,24 +167,24 @@ func NewCluster(scheduler *scheduler.Scheduler, TLSConfig *tls.Config, master st
 
 // Handle callbacks for the events
 func (c *Cluster) Handle(e *cluster.Event) error {
-	// call Handle for all eventHandlers
-	c.eventHandlers.Handle(e)
+	// call Handle for all clusterEventHandlers
+	c.clusterEventHandlers.Handle(e)
 	return nil
 }
 
 // RegisterEventHandler registers an event handler.
 func (c *Cluster) RegisterEventHandler(h cluster.EventHandler) error {
-	return c.eventHandlers.RegisterEventHandler(h)
+	return c.clusterEventHandlers.RegisterEventHandler(h)
 }
 
 // UnregisterEventHandler unregisters a previously registered event handler.
 func (c *Cluster) UnregisterEventHandler(h cluster.EventHandler) {
-	c.eventHandlers.UnregisterEventHandler(h)
+	c.clusterEventHandlers.UnregisterEventHandler(h)
 }
 
-// NewAPIEventsHandler creates a new API events handler
-func (c *Cluster) NewAPIEventsHandler() *cluster.EventsHandler {
-	return cluster.NewEventsHandler()
+// NewAPIEventHandler creates a new API events handler
+func (c *Cluster) NewAPIEventHandler() *cluster.APIEventHandler {
+	return cluster.NewAPIEventHandler()
 }
 
 // CloseWatchQueue closes the watchQueue when the manager shuts down.