Add containers update agent service (#187)

[#186] Implement containers update agent service

Merge update-agent feature branch
* Extend daemon configuration and dummy implementation for update agent (#168)
* Upgrade google.golang.org/grpc to version 1.53.0 or later (#172)
* Preparation for implementing apply of containers desired state (#175)
* Implement apply of desired state for containers update agent (#177)
* Implement current state reporting for containers update agent (#179)

Signed-off-by: Kristiyan Gostev <[email protected]>
Co-authored-by: Dimitar Dimitrov <[email protected]>
Co-authored-by: Stoyan Zoubev <[email protected]>

Author: 3 people

containerm/cli/cli_commamd_ctrs_base_test.go

@@ -122,7 +122,12 @@ func execTestsRun(t *testing.T, cliTest cliCommandTest) {
 			// perform the real call
 			resultErr := cliTest.runCommand(testCase.args)
 			// assert result
-			testutil.AssertError(t, expectedRunErr, resultErr)
+			if expectedRunErr == nil {
+				testutil.AssertNil(t, resultErr)
+			} else {
+				testutil.AssertNotNil(t, resultErr)
+				testutil.AssertContainsString(t, resultErr.Error(), expectedRunErr.Error())
+			}
 		})
 	}
 }

containerm/cli/cli_command_ctrs_create.go

@@ -15,9 +15,6 @@ package main
 import (
 	"context"
 	"fmt"
-	"net"
-	"strconv"
-	"strings"
 	"time"
 
 	"github.com/eclipse-kanto/container-management/containerm/containers/types"
@@ -119,23 +116,23 @@ func (cc *createCmd) run(args []string) error {
 	}
 
 	if cc.config.devices != nil {
-		devs, err := parseDevices(cc.config.devices)
+		devs, err := util.ParseDeviceMappings(cc.config.devices)
 		if err != nil {
 			return err
 		}
 		ctrToCreate.HostConfig.Devices = devs
 	}
 
 	if cc.config.mountPoints != nil {
-		mounts, err := parseMountPoints(cc.config.mountPoints)
+		mounts, err := util.ParseMountPoints(cc.config.mountPoints)
 		if err != nil {
 			return err
 		} else if mounts != nil {
 			ctrToCreate.Mounts = mounts
 		}
 	}
 	if cc.config.ports != nil {
-		mappings, err := parsePortMappings(cc.config.ports)
+		mappings, err := util.ParsePortMappings(cc.config.ports)
 		if err != nil {
 			return err
 		}
@@ -310,122 +307,3 @@ func (cc *createCmd) setupFlags() {
 	flagSet.StringSliceVar(&cc.config.decKeys, "dec-keys", nil, "Sets a list of private keys filenames (GPG private key ring, JWE and PKCS7 private key). Each entry can include an optional password separated by a colon after the filename.")
 	flagSet.StringSliceVar(&cc.config.decRecipients, "dec-recipients", nil, "Sets a recipients certificates list of the image (used only for PKCS7 and must be an x509)")
 }
-
-func parseDevices(devices []string) ([]types.DeviceMapping, error) {
-	var devs []types.DeviceMapping
-	for _, devPair := range devices {
-		pair := strings.Split(strings.TrimSpace(devPair), ":")
-		if len(pair) == 2 {
-			devs = append(devs, types.DeviceMapping{
-				PathOnHost:        pair[0],
-				PathInContainer:   pair[1],
-				CgroupPermissions: "rwm",
-			})
-		} else if len(pair) == 3 {
-			if len(pair[2]) == 0 || len(pair[2]) > 3 {
-				return nil, log.NewError("incorrect device cgroup permissions format")
-			}
-			for i := 0; i < len(pair[2]); i++ {
-				if (pair[2])[i] != "w"[0] && (pair[2])[i] != "r"[0] && (pair[2])[i] != "m"[0] {
-					return nil, log.NewError("incorrect device cgroup permissions format")
-				}
-			}
-
-			devs = append(devs, types.DeviceMapping{
-				PathOnHost:        pair[0],
-				PathInContainer:   pair[1],
-				CgroupPermissions: pair[2],
-			})
-		} else {
-			return nil, log.NewError("incorrect device configuration format")
-		}
-	}
-	return devs, nil
-}
-
-func parseMountPoints(mps []string) ([]types.MountPoint, error) {
-	var mountPoints []types.MountPoint
-	var mountPoint types.MountPoint
-	for _, mp := range mps {
-		mount := strings.Split(strings.TrimSpace(mp), ":")
-		// if propagation mode is omitted, "rprivate" is set as default
-		if len(mount) < 2 || len(mount) > 3 {
-			return nil, log.NewError("Incorrect number of parameters of the mount point")
-		}
-		mountPoint = types.MountPoint{
-			Destination: mount[1],
-			Source:      mount[0],
-		}
-		if len(mount) == 2 {
-			log.Debug("propagation mode ommited - setting default to rprivate")
-			mountPoint.PropagationMode = types.RPrivatePropagationMode
-		} else {
-			mountPoint.PropagationMode = mount[2]
-		}
-		mountPoints = append(mountPoints, mountPoint)
-	}
-	return mountPoints, nil
-}
-
-func parsePortMappings(mappings []string) ([]types.PortMapping, error) {
-	var (
-		portMappings  []types.PortMapping
-		err           error
-		protocol      string
-		containerPort int64
-		hostIP        string
-		hostPort      int64
-		hostPortEnd   int64
-	)
-
-	for _, mapping := range mappings {
-		mappingWithProto := strings.Split(strings.TrimSpace(mapping), "/")
-		mapping = mappingWithProto[0]
-		if len(mappingWithProto) == 2 {
-			// port is specified, e.g.80:80/tcp
-			protocol = mappingWithProto[1]
-		}
-		addressAndPorts := strings.Split(strings.TrimSpace(mapping), ":")
-		hostPortIdx := 0 // if host ip not set
-		if len(addressAndPorts) == 2 {
-			// host address not specified, e.g. 80:80
-		} else if len(addressAndPorts) == 3 {
-			hostPortIdx = 1
-			hostIP = addressAndPorts[0]
-			validIP := net.ParseIP(hostIP)
-			if validIP == nil {
-				return nil, log.NewError("Incorrect host ip port mapping configuration")
-			}
-			hostPort, err = strconv.ParseInt(addressAndPorts[1], 10, 32)
-			containerPort, err = strconv.ParseInt(addressAndPorts[2], 10, 32)
-
-		} else {
-			return nil, log.NewError("Incorrect port mapping configuration")
-		}
-
-		hostPortWithRange := strings.Split(strings.TrimSpace(addressAndPorts[hostPortIdx]), "-")
-		if len(hostPortWithRange) == 2 {
-			hostPortEnd, err = strconv.ParseInt(hostPortWithRange[1], 10, 32)
-			if err != nil {
-				return nil, log.NewError("Incorrect host range port mapping configuration")
-			}
-			hostPort, err = strconv.ParseInt(hostPortWithRange[0], 10, 32)
-		} else {
-			hostPort, err = strconv.ParseInt(addressAndPorts[hostPortIdx], 10, 32)
-		}
-		containerPort, err = strconv.ParseInt(addressAndPorts[hostPortIdx+1], 10, 32)
-		if err != nil {
-			return nil, log.NewError("Incorrect port mapping configuration, parsing error")
-		}
-
-		portMappings = append(portMappings, types.PortMapping{
-			Proto:         protocol,
-			ContainerPort: uint16(containerPort),
-			HostIP:        hostIP,
-			HostPort:      uint16(hostPort),
-			HostPortEnd:   uint16(hostPortEnd),
-		})
-	}
-	return portMappings, nil
-
-}

containerm/cli/cli_command_ctrs_create_test.go

@@ -701,12 +701,12 @@ func (createTc *createCommandTest) mockExecCreateDevicesWithPrivileged(args []st
 
 func (createTc *createCommandTest) mockExecCreateDevicesErrConfigFormat(args []string) error {
 	createTc.mockClient.EXPECT().Create(gomock.AssignableToTypeOf(context.Background()), gomock.Any()).Times(0)
-	return log.NewError("incorrect device configuration format")
+	return log.NewError("incorrect configuration value for device mapping")
 }
 
 func (createTc *createCommandTest) mockExecCreateDevicesErrCgroupFormat(args []string) error {
 	createTc.mockClient.EXPECT().Create(gomock.AssignableToTypeOf(context.Background()), gomock.Any()).Times(0)
-	return log.NewError("incorrect device cgroup permissions format")
+	return log.NewError("incorrect cgroup permissions format for device mapping")
 }
 
 func (createTc *createCommandTest) mockExecCreateWithMountPoints(args []string) error {
@@ -878,7 +878,7 @@ func (createTc *createCommandTest) mockExecCreateWithPortsIncorrectPortsConfig(a
 
 func (createTc *createCommandTest) mockExecCreateWithPortsIncorrectPortsConfigParseErr(args []string) error {
 	createTc.mockClient.EXPECT().Create(gomock.AssignableToTypeOf(context.Background()), gomock.Any()).Times(0)
-	return log.NewError("Incorrect port mapping configuration, parsing error")
+	return log.NewError("Incorrect container port mapping configuration")
 }
 
 func (createTc *createCommandTest) mockExecCreateWithPortsIncorrectHostRange(args []string) error {

containerm/daemon/daemon_command.go

@@ -85,6 +85,12 @@ func setupCommandFlags(cmd *cobra.Command) {
 	flagSet.StringVar(&cfg.ThingsConfig.ThingsMetaPath, "things-home-dir", cfg.ThingsConfig.ThingsMetaPath, "Specify the home directory for the things container management service persistent storage")
 	flagSet.StringSliceVar(&cfg.ThingsConfig.Features, "things-features", cfg.ThingsConfig.Features, "Specify the desired Ditto features that will be registered for the containers Ditto thing")
 
+	// init update agent
+	flagSet.BoolVar(&cfg.UpdateAgentConfig.UpdateAgentEnable, "ua-enable", cfg.UpdateAgentConfig.UpdateAgentEnable, "Enable the update agent for containers")
+	flagSet.StringVar(&cfg.UpdateAgentConfig.DomainName, "ua-domain", cfg.UpdateAgentConfig.DomainName, "Specify the domain name for the containers update agent")
+	flagSet.StringSliceVar(&cfg.UpdateAgentConfig.SystemContainers, "ua-system-containers", cfg.UpdateAgentConfig.SystemContainers, "Specify the list of system containers which shall be skipped during update process by the update agent")
+	flagSet.BoolVar(&cfg.UpdateAgentConfig.VerboseInventoryReport, "ua-verbose-inventory-report", cfg.UpdateAgentConfig.VerboseInventoryReport, "Enables verbose reporting of current inventory of containers by the update agent")
+
 	// init local communication flags
 	flagSet.StringVar(&cfg.LocalConnection.BrokerURL, "conn-broker-url", cfg.LocalConnection.BrokerURL, "Specify the MQTT broker URL to connect to")
 	flagSet.StringVar(&cfg.LocalConnection.KeepAlive, "conn-keep-alive", cfg.LocalConnection.KeepAlive, "Specify the keep alive duration for the MQTT requests as duration string")

containerm/daemon/daemon_config.go

@@ -35,6 +35,8 @@ type config struct {
 
 	ThingsConfig *thingsConfig `json:"things,omitempty"`
 
+	UpdateAgentConfig *updateAgentConfig `json:"update_agent,omitempty"`
+
 	LocalConnection *localConnectionConfig `json:"connection,omitempty"`
 }
 
@@ -151,6 +153,14 @@ type thingsConfig struct {
 	ThingsConnectionConfig *thingsConnectionConfig `json:"connection,omitempty"`
 }
 
+// things client configuration
+type updateAgentConfig struct {
+	UpdateAgentEnable      bool     `json:"enable,omitempty"`
+	DomainName             string   `json:"domain,omitempty"`
+	SystemContainers       []string `json:"system_containers,omitempty"`
+	VerboseInventoryReport bool     `json:"verbose_inventory_report,omitempty"`
+}
+
 // local connection config
 type localConnectionConfig struct {
 	BrokerURL          string     `json:"broker_url,omitempty"`

containerm/daemon/daemon_config_default.go

@@ -95,6 +95,11 @@ const (
 	deploymentModeDefault     = string(deployment.UpdateMode)
 	deploymentMetaPathDefault = managerMetaPathDefault
 	deploymentCtrPathDefault  = "/etc/container-management/containers"
+
+	// default update agent config
+	updateAgentEnableDefault                 = false
+	updateAgentDomainDefault                 = "containers"
+	updateAgentVerboseInventoryReportDefault = false
 )
 
 var (
@@ -178,6 +183,12 @@ func getDefaultInstance() *config {
 			DeploymentMetaPath: deploymentMetaPathDefault,
 			DeploymentCtrPath:  deploymentCtrPathDefault,
 		},
+		UpdateAgentConfig: &updateAgentConfig{
+			UpdateAgentEnable:      updateAgentEnableDefault,
+			DomainName:             updateAgentDomainDefault,
+			SystemContainers:       []string{}, // no system containers by defaults
+			VerboseInventoryReport: updateAgentVerboseInventoryReportDefault,
+		},
 		LocalConnection: &localConnectionConfig{
 			BrokerURL:          connectionBrokerURLDefault,
 			KeepAlive:          connectionKeepAliveDefault,

containerm/daemon/daemon_config_util.go

@@ -28,6 +28,7 @@ import (
 	"github.com/eclipse-kanto/container-management/containerm/network"
 	"github.com/eclipse-kanto/container-management/containerm/server"
 	"github.com/eclipse-kanto/container-management/containerm/things"
+	"github.com/eclipse-kanto/container-management/containerm/updateagent"
 	"github.com/spf13/pflag"
 )
 
@@ -117,15 +118,6 @@ func extractThingsOptions(daemonConfig *config) []things.ContainerThingsManagerO
 		}
 	}
 
-	parseDuration := func(duration, defaultDuration string) time.Duration {
-		d, err := time.ParseDuration(duration)
-		if err != nil {
-			log.Warn("Invalid Duration string: %s", duration)
-			d, _ = time.ParseDuration(defaultDuration)
-		}
-		return d
-	}
-
 	thingsOpts = append(thingsOpts,
 		things.WithMetaPath(daemonConfig.ThingsConfig.ThingsMetaPath),
 		things.WithFeatures(daemonConfig.ThingsConfig.Features),
@@ -145,6 +137,30 @@ func extractThingsOptions(daemonConfig *config) []things.ContainerThingsManagerO
 	return thingsOpts
 }
 
+func extractUpdateAgentOptions(daemonConfig *config) []updateagent.ContainersUpdateAgentOpt {
+	updateAgentOpts := []updateagent.ContainersUpdateAgentOpt{}
+	updateAgentOpts = append(updateAgentOpts,
+		updateagent.WithDomainName(daemonConfig.UpdateAgentConfig.DomainName),
+		updateagent.WithSystemContainers(daemonConfig.UpdateAgentConfig.SystemContainers),
+		updateagent.WithVerboseInventoryReport(daemonConfig.UpdateAgentConfig.VerboseInventoryReport),
+
+		updateagent.WithConnectionBroker(daemonConfig.LocalConnection.BrokerURL),
+		updateagent.WithConnectionKeepAlive(parseDuration(daemonConfig.LocalConnection.KeepAlive, connectionKeepAliveDefault)),
+		updateagent.WithConnectionDisconnectTimeout(parseDuration(daemonConfig.LocalConnection.DisconnectTimeout, connectionDisconnectTimeoutDefault)),
+		updateagent.WithConnectionClientUsername(daemonConfig.LocalConnection.ClientUsername),
+		updateagent.WithConnectionClientPassword(daemonConfig.LocalConnection.ClientPassword),
+		updateagent.WithConnectionConnectTimeout(parseDuration(daemonConfig.LocalConnection.ConnectTimeout, connectTimeoutTimeoutDefault)),
+		updateagent.WithConnectionAcknowledgeTimeout(parseDuration(daemonConfig.LocalConnection.AcknowledgeTimeout, acknowledgeTimeoutDefault)),
+		updateagent.WithConnectionSubscribeTimeout(parseDuration(daemonConfig.LocalConnection.SubscribeTimeout, subscribeTimeoutDefault)),
+		updateagent.WithConnectionUnsubscribeTimeout(parseDuration(daemonConfig.LocalConnection.UnsubscribeTimeout, unsubscribeTimeoutDefault)),
+	)
+	transport := daemonConfig.LocalConnection.Transport
+	if transport != nil {
+		updateAgentOpts = append(updateAgentOpts, updateagent.WithTLSConfig(transport.RootCA, transport.ClientCert, transport.ClientKey))
+	}
+	return updateAgentOpts
+}
+
 func extractDeploymentMgrOptions(daemonConfig *config) []deployment.Opt {
 	return []deployment.Opt{
 		deployment.WithMode(daemonConfig.DeploymentManagerConfig.DeploymentMode),
@@ -218,6 +234,9 @@ func dumpConfiguration(configInstance *config) {
 	// dump things client config
 	dumpThingsClient(configInstance)
 
+	// dump update agent config
+	dumpUpdateAgent(configInstance)
+
 	// dump deployment manager config
 	dumpDeploymentManager(configInstance)
 
@@ -335,6 +354,17 @@ func dumpThingsClient(configInstance *config) {
 	}
 }
 
+func dumpUpdateAgent(configInstance *config) {
+	if configInstance.UpdateAgentConfig != nil {
+		log.Debug("[daemon_cfg][ua-enable] : %v", configInstance.UpdateAgentConfig.UpdateAgentEnable)
+		if configInstance.UpdateAgentConfig.UpdateAgentEnable {
+			log.Debug("[daemon_cfg][ua-domain] : %s", configInstance.UpdateAgentConfig.DomainName)
+			log.Debug("[daemon_cfg][ua-system-containers] : %s", configInstance.UpdateAgentConfig.SystemContainers)
+			log.Debug("[daemon_cfg][ua-verbose-inventory-report] : %v", configInstance.UpdateAgentConfig.VerboseInventoryReport)
+		}
+	}
+}
+
 func dumpDeploymentManager(configInstance *config) {
 	if configInstance.DeploymentManagerConfig != nil {
 		log.Debug("[daemon_cfg][deployment-enable] : %v", configInstance.DeploymentManagerConfig.DeploymentEnable)
@@ -428,3 +458,12 @@ func applyInsecureRegistryConfig(registriesConfig map[string]*ctr.RegistryConfig
 	}
 	return res
 }
+
+func parseDuration(duration, defaultDuration string) time.Duration {
+	d, err := time.ParseDuration(duration)
+	if err != nil {
+		log.Warn("Invalid Duration string: %s", duration)
+		d, _ = time.ParseDuration(defaultDuration)
+	}
+	return d
+}