feat: implement service events

This implements service events, adds test for events API based on
service events as they're the easiest to generate on demand.

Disabled validate test for 'metal' as it validates disk device against
local system which doesn't make much sense.

Signed-off-by: Andrey Smirnov <smirnov.andrey@gmail.com>

api/machine/machine.proto

@@ -77,6 +77,22 @@ message TaskEvent {
   Action action = 2;
 }
 
+message ServiceStateEvent {
+  string service = 1;
+  enum Action {
+    INITIALIZED = 0;
+    PREPARING = 1;
+    WAITING = 2;
+    RUNNING = 3;
+    STOPPING = 4;
+    FINISHED = 5;
+    FAILED = 6;
+    SKIPPED = 7;
+  };
+  Action action = 2;
+  string message = 3;
+};
+
 message EventsRequest {}
 
 message Event {

cmd/talosctl/cmd/talos/events.go

@@ -53,6 +53,8 @@ var eventsCmd = &cobra.Command{
 						} else {
 							args = []interface{}{msg.GetSequence() + " " + msg.GetAction().String()}
 						}
+					case *machine.ServiceStateEvent:
+						args = []interface{}{fmt.Sprintf("%s [%s]: %s", msg.GetService(), msg.GetAction(), msg.GetMessage())}
 					default:
 						// We haven't implemented the handling of this event yet.
 						continue

go.mod

@@ -74,7 +74,7 @@ require (
 	golang.org/x/text v0.3.2
 	golang.org/x/time v0.0.0-20191024005414-555d28b269f0
 	google.golang.org/grpc v1.27.0
-	google.golang.org/protobuf v1.25.0 // indirect
+	google.golang.org/protobuf v1.25.0
 	gopkg.in/freddierice/go-losetup.v1 v1.0.0-20170407175016-fc9adea44124
 	gopkg.in/fsnotify.v1 v1.4.7
 	gopkg.in/yaml.v2 v2.2.8

internal/app/machined/pkg/system/events/events.go

@@ -60,6 +60,15 @@ type ServiceEvent struct {
 	Timestamp time.Time
 }
 
+// AsProto returns protobuf representation of respective machined event.
+func (event *ServiceEvent) AsProto(service string) *machineapi.ServiceStateEvent {
+	return &machineapi.ServiceStateEvent{
+		Service: service,
+		Action:  machineapi.ServiceStateEvent_Action(event.State),
+		Message: event.Message,
+	}
+}
+
 // ServiceEvents is a fixed length history of events.
 type ServiceEvents struct {
 	events    []ServiceEvent

internal/app/machined/pkg/system/service_runner.go

@@ -88,6 +88,10 @@ func (svcrunner *ServiceRunner) UpdateState(newstate events.ServiceState, messag
 	isFinished := svcrunner.inStateLocked(StateEventFinished)
 	svcrunner.mu.Unlock()
 
+	if svcrunner.runtime != nil {
+		svcrunner.runtime.Events().Publish(event.AsProto(svcrunner.id))
+	}
+
 	if isUp {
 		svcrunner.notifyEvent(StateEventUp)
 	}

internal/integration/api/events.go

@@ -0,0 +1,104 @@
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+// +build integration_api
+
+package api
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/talos-systems/talos/api/machine"
+	"github.com/talos-systems/talos/internal/integration/base"
+	"github.com/talos-systems/talos/pkg/client"
+)
+
+// EventsSuite verifies Events API.
+type EventsSuite struct {
+	base.APISuite
+
+	ctx       context.Context
+	ctxCancel context.CancelFunc
+}
+
+// SuiteName ...
+func (suite *EventsSuite) SuiteName() string {
+	return "api.EventsSuite"
+}
+
+// SetupTest ...
+func (suite *EventsSuite) SetupTest() {
+	// make sure API calls have timeout
+	suite.ctx, suite.ctxCancel = context.WithTimeout(context.Background(), 2*time.Minute)
+}
+
+// TearDownTest ...
+func (suite *EventsSuite) TearDownTest() {
+	suite.ctxCancel()
+}
+
+// TestServiceEvents verifies that service restart generates events.
+func (suite *EventsSuite) TestServiceEvents() {
+	const service = "timed" // any restartable service should work
+
+	ctx, ctxCancel := context.WithTimeout(suite.ctx, 30*time.Second)
+	defer ctxCancel()
+
+	svcInfo, err := suite.Client.ServiceInfo(ctx, service)
+	suite.Require().NoError(err)
+
+	if len(svcInfo) == 0 { // service is not registered (e.g. docker)
+		suite.T().Skip(fmt.Sprintf("skipping test as service %q is not registered", service))
+	}
+
+	actionsSeen := make(map[machine.ServiceStateEvent_Action]struct{})
+
+	checkExpectedActions := func() error {
+		for _, action := range []machine.ServiceStateEvent_Action{
+			machine.ServiceStateEvent_STOPPING,
+			machine.ServiceStateEvent_FINISHED,
+			machine.ServiceStateEvent_WAITING,
+			machine.ServiceStateEvent_PREPARING,
+			machine.ServiceStateEvent_RUNNING,
+		} {
+			if _, ok := actionsSeen[action]; !ok {
+				return fmt.Errorf("expected action %s was not seen", action)
+			}
+		}
+
+		return nil
+	}
+
+	go func() {
+		suite.Assert().NoError(suite.Client.EventsWatch(ctx, func(ch <-chan client.Event) {
+			defer ctxCancel()
+
+			for event := range ch {
+				if msg, ok := event.Payload.(*machine.ServiceStateEvent); ok && msg.GetService() == service {
+					actionsSeen[msg.GetAction()] = struct{}{}
+				}
+
+				if checkExpectedActions() == nil {
+					return
+				}
+			}
+		}))
+	}()
+
+	// wait for event watcher to start
+	time.Sleep(200 * time.Millisecond)
+
+	_, err = suite.Client.ServiceRestart(ctx, service)
+	suite.Assert().NoError(err)
+
+	<-ctx.Done()
+
+	suite.Require().NoError(checkExpectedActions())
+}
+
+func init() {
+	allSuites = append(allSuites, new(EventsSuite))
+}

internal/integration/cli/validate.go

@@ -7,6 +7,7 @@
 package cli
 
 import (
+	"fmt"
 	"io/ioutil"
 	"os"
 
@@ -47,8 +48,10 @@ func (suite *ValidateSuite) TestValidate() {
 	suite.RunCLI([]string{"gen", "config", "foobar", "https://10.0.0.1"})
 
 	for _, configFile := range []string{"init.yaml", "controlplane.yaml", "join.yaml"} {
-		for _, mode := range []string{"cloud", "container", "metal"} {
+		for _, mode := range []string{"cloud", "container"} {
-			suite.RunCLI([]string{"validate", "-m", mode, "-c", configFile})
+			suite.Run(fmt.Sprintf("%s-%s", configFile, mode), func() {
+				suite.RunCLI([]string{"validate", "-m", mode, "-c", configFile})
+			})
 		}
 	}
 }

pkg/client/client.go

@@ -806,21 +806,26 @@ func (c *Client) EventsWatch(ctx context.Context, watchFunc func(<-chan Event))
 
 		var msg proto.Message
 
-		seqEvent := &machineapi.SequenceEvent{}
+		for _, eventType := range []proto.Message{
-
+			&machineapi.SequenceEvent{},
-		switch typeURL {
+			&machineapi.ServiceStateEvent{},
-		case "talos/runtime/" + string(seqEvent.ProtoReflect().Descriptor().FullName()):
+		} {
-			msg = &machineapi.SequenceEvent{}
+			if typeURL == "talos/runtime/"+string(eventType.ProtoReflect().Descriptor().FullName()) {
-
+				msg = eventType
-			if err = proto.Unmarshal(event.GetData().GetValue(), msg); err != nil {
+				break
-				log.Printf("failed to unmarshal message: %v", err) // TODO: this should be fixed to return errors
-				continue
 			}
-		default:
+		}
+
+		if msg == nil {
 			// We haven't implemented the handling of this event yet.
 			continue
 		}
 
+		if err = proto.Unmarshal(event.GetData().GetValue(), msg); err != nil {
+			log.Printf("failed to unmarshal message: %v", err) // TODO: this should be fixed to return errors
+			continue
+		}
+
 		ev := Event{
 			Node:    defaultNode,
 			TypeURL: typeURL,