Add file watcher for Appnet agent image update

Author: mythri-garaga

agent/engine/docker_task_engine.go

@@ -111,7 +111,11 @@ const (
 	stopContainerMaxRetryCount     = 5
 )
 
-var newExponentialBackoff = retry.NewExponentialBackoff
+var (
+	newExponentialBackoff = retry.NewExponentialBackoff
+
+	serviceConnectAppnetAgenTarballDir = "/var/lib/ecs/deps/serviceconnect/"
+)
 
 // DockerTaskEngine is a state machine for managing a task and its containers
 // in ECS.
@@ -281,6 +285,7 @@ func (engine *DockerTaskEngine) Init(ctx context.Context) error {
 	go engine.handleDockerEvents(derivedCtx)
 	engine.initialized = true
 	go engine.startPeriodicExecAgentsMonitoring(derivedCtx)
+	go engine.watchAppNetImage(engine.ctx, serviceConnectAppnetAgenTarballDir)
 	return nil
 }
 

agent/engine/docker_task_engine_linux.go

@@ -17,10 +17,15 @@
 package engine
 
 import (
+	"context"
 	"time"
 
+	"github.com/cihub/seelog"
+	"github.com/fsnotify/fsnotify"
+
 	apicontainer "github.com/aws/amazon-ecs-agent/agent/api/container"
 	apitask "github.com/aws/amazon-ecs-agent/agent/api/task"
+	apitaskstatus "github.com/aws/amazon-ecs-agent/agent/api/task/status"
 )
 
 const (
@@ -39,3 +44,78 @@ func (engine *DockerTaskEngine) updateTaskENIDependencies(task *apitask.Task) {
 func (engine *DockerTaskEngine) invokePluginsForContainer(task *apitask.Task, container *apicontainer.Container) error {
 	return nil
 }
+
+func (engine *DockerTaskEngine) watchAppNetImage(ctx context.Context, filepath string) {
+	watcher, err := fsnotify.NewWatcher()
+	if err != nil {
+		seelog.Errorf("error adding %s to fsnotify watcher, error: %v", filepath, err)
+	}
+	err = watcher.Add(filepath)
+	if err != nil {
+		seelog.Errorf("failed to initialize fsnotify NewWatcher, error: %v", err)
+	}
+	defer watcher.Close()
+	engine.watch(ctx, watcher)
+}
+
+func (engine *DockerTaskEngine) watch(ctx context.Context, watcher *fsnotify.Watcher) {
+	// Start listening for events.
+	for {
+		select {
+		case event, ok := <-watcher.Events:
+			if !ok {
+				seelog.Warn("fsnotify watcher event error")
+				return
+			}
+			const writeOrCreateMask = fsnotify.Write | fsnotify.Create
+			if event.Op&writeOrCreateMask != 0 {
+				seelog.Debugf("new fsnotify watcher event: %s", event.Name)
+				// reload the updated Appnet Agent image
+				if err := engine.reloadAppNetImage(); err != nil {
+					// restart the internal instance relay task with
+					// updated Appnet Agent image
+					engine.restartInstanceTask()
+				}
+			}
+		case err, ok := <-watcher.Errors:
+			if !ok {
+				return
+			}
+			seelog.Error("fsnotify watcher error: %v", err)
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+func (engine *DockerTaskEngine) reloadAppNetImage() error {
+	_, err := engine.serviceconnectManager.LoadImage(engine.ctx, engine.cfg, engine.client)
+	if err != nil {
+		seelog.Errorf("engine: Failed to reload appnet Agent container, error: %v", err)
+		return err
+	}
+	return nil
+}
+
+func (engine *DockerTaskEngine) restartInstanceTask() {
+	if engine.serviceconnectRelay != nil {
+		serviceconnectRelayTask, err := engine.serviceconnectManager.CreateInstanceTask(engine.cfg)
+		if err != nil {
+			seelog.Errorf("Unable to start relay for task in the engine: %v", err)
+			return
+		}
+		// clean up instance relay task
+		for _, container := range engine.serviceconnectRelay.Containers {
+			if container.Type == apicontainer.ContainerServiceConnectRelay {
+				engine.stopContainer(engine.serviceconnectRelay, container)
+			}
+		}
+		engine.serviceconnectRelay.SetDesiredStatus(apitaskstatus.TaskStopped)
+		engine.sweepTask(engine.serviceconnectRelay)
+		engine.deleteTask(engine.serviceconnectRelay)
+
+		engine.serviceconnectRelay = serviceconnectRelayTask
+		engine.AddTask(engine.serviceconnectRelay)
+		seelog.Info("engine: Restarted AppNet Relay task")
+	}
+}

agent/engine/docker_task_engine_linux_test.go

@@ -26,6 +26,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/fsnotify/fsnotify"
+
 	"github.com/aws/amazon-ecs-agent/agent/api/appmesh"
 	apicontainer "github.com/aws/amazon-ecs-agent/agent/api/container"
 	apicontainerstatus "github.com/aws/amazon-ecs-agent/agent/api/container/status"
@@ -1364,3 +1366,29 @@ func TestProvisionContainerResourcesBridgeModeWithServiceConnect(t *testing.T) {
 		require.Nil(t, taskEngine.(*DockerTaskEngine).provisionContainerResources(testTask, cont).Error)
 	}
 }
+
+func TestWatchAppNetImage(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.TODO())
+	defer cancel()
+	ctrl, _, _, taskEngine, _, _, _, serviceConnectManager := mocks(t, ctx, &defaultConfig)
+	defer ctrl.Finish()
+
+	// create a fsnotify watcher
+	watcher, err := fsnotify.NewWatcher()
+	assert.NoError(t, err)
+	err = watcher.Add("/tmp/")
+	assert.NoError(t, err)
+	defer watcher.Close()
+	event := fsnotify.Event{
+		Name: "/tmp/",
+		Op:   fsnotify.Create,
+	}
+	serviceConnectManager.EXPECT().LoadImage(gomock.Any(), gomock.Any(), gomock.Any()).Times(1)
+
+	watcherCtx, watcherCancel := context.WithTimeout(context.Background(), time.Second)
+	defer watcherCancel()
+	go taskEngine.(*DockerTaskEngine).watch(watcherCtx, watcher)
+	watcher.Events <- event
+
+	<-watcherCtx.Done()
+}

agent/engine/docker_task_engine_unsupported.go

@@ -17,10 +17,12 @@
 package engine
 
 import (
+	"context"
 	"time"
 
 	apicontainer "github.com/aws/amazon-ecs-agent/agent/api/container"
 	apitask "github.com/aws/amazon-ecs-agent/agent/api/task"
+	"github.com/fsnotify/fsnotify"
 )
 
 const (
@@ -39,3 +41,21 @@ func (engine *DockerTaskEngine) updateTaskENIDependencies(task *apitask.Task) {
 func (engine *DockerTaskEngine) invokePluginsForContainer(task *apitask.Task, container *apicontainer.Container) error {
 	return nil
 }
+
+// watchAppNetImage is a file watcher, if there is any change/update to AppNet image
+// we reload the image and restart the relay instance task with updated AppNet image.
+func (engine *DockerTaskEngine) watchAppNetImage(ctx context.Context, filepath string) {
+}
+
+func (engine *DockerTaskEngine) watch(ctx context.Context, watcher *fsnotify.Watcher) {
+}
+
+// reloadAppNetImage reloads the new AppNet image for service connect
+func (engine *DockerTaskEngine) reloadAppNetImage() error {
+	return nil
+}
+
+// restartInstanceTask stop the running internal relay task and starts a new one
+// with updated AppNet image
+func (engine *DockerTaskEngine) restartInstanceTask() {
+}

agent/engine/docker_task_engine_windows.go

@@ -17,8 +17,11 @@
 package engine
 
 import (
+	"context"
 	"time"
 
+	"github.com/fsnotify/fsnotify"
+
 	apicontainer "github.com/aws/amazon-ecs-agent/agent/api/container"
 	apitask "github.com/aws/amazon-ecs-agent/agent/api/task"
 	"github.com/aws/amazon-ecs-agent/agent/logger"
@@ -65,3 +68,16 @@ func (engine *DockerTaskEngine) invokePluginsForContainer(task *apitask.Task, co
 
 	return nil
 }
+
+func (engine *DockerTaskEngine) watchAppNetImage(ctx context.Context, filepath string) {
+}
+
+func (engine *DockerTaskEngine) watch(ctx context.Context, watcher *fsnotify.Watcher) {
+}
+
+func (engine *DockerTaskEngine) reloadAppNetImage() error {
+	return nil
+}
+
+func (engine *DockerTaskEngine) restartInstanceTask() {
+}

agent/engine/serviceconnect/manager_linux.go

@@ -387,17 +387,17 @@ func (agent *manager) LoadImage(ctx context.Context, _ *config.Config, dockerCli
 			})
 			continue
 		}
-		logger.Debug(fmt.Sprintf("Loading appnet agent container tarball: %s", agentContainerTarballPath))
+		logger.Debug(fmt.Sprintf("Loading Appnet agent container tarball: %s", agentContainerTarballPath))
 		if loadErr = loader.LoadFromFile(ctx, agentContainerTarballPath, dockerClient); loadErr != nil {
-			logger.Warn(fmt.Sprintf("Unable to load appnet agent container tarball: %s", agentContainerTarballPath),
+			logger.Warn(fmt.Sprintf("Unable to load Appnet agent container tarball: %s", agentContainerTarballPath),
 				logger.Fields{
 					field.Error: loadErr,
 				})
 			continue
 		}
 		agent.setLoadedAppnetVerion(supportedAppnetInterfaceVersion)
 		imageName, _ := agent.GetLoadedImageName()
-		logger.Info(fmt.Sprintf("Successfully loaded appnet agent container tarball: %s", agentContainerTarballPath),
+		logger.Info(fmt.Sprintf("Successfully loaded Appnet agent container tarball: %s", agentContainerTarballPath),
 			logger.Fields{
 				field.Image: imageName,
 			})

agent/go.mod

@@ -18,6 +18,7 @@ require (
 	github.com/docker/docker v0.0.0-20200531234253-77e06fda0c94
 	github.com/docker/go-connections v0.4.0
 	github.com/docker/go-units v0.4.0
+	github.com/fsnotify/fsnotify v1.5.4
 	github.com/godbus/dbus/v5 v5.0.6 // indirect
 	github.com/golang/mock v1.1.1
 	github.com/google/go-cmp v0.5.6 // indirect
@@ -34,13 +35,13 @@ require (
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v0.9.4
 	github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4
-	github.com/prometheus/procfs v0.6.0 // indirect
 	github.com/prometheus/common v0.4.1
+	github.com/prometheus/procfs v0.6.0 // indirect
 	github.com/stretchr/testify v1.7.0
 	github.com/vishvananda/netlink v1.1.0
 	go.etcd.io/bbolt v1.3.6
 	golang.org/x/net v0.0.0-20210525063256-abc453219eb5
-	golang.org/x/sys v0.0.0-20210510120138-977fb7262007
+	golang.org/x/sys v0.0.0-20220412211240-33da011f77ad
 	golang.org/x/text v0.3.6 // indirect
 	golang.org/x/time v0.0.0-20170927054726-6dc17368e09b // indirect
 	golang.org/x/tools v0.1.5

agent/go.sum

@@ -66,6 +66,8 @@ github.com/envoyproxy/go-control-plane v0.9.9-0.20210217033140-668b12f5399d/go.m
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
 github.com/frankban/quicktest v1.11.3 h1:8sXhOn0uLys67V8EsXLc6eszDs8VXWxL3iRvebPhedY=
 github.com/frankban/quicktest v1.11.3/go.mod h1:wRf/ReqHper53s+kmmSZizM8NamnL3IM0I9ntUbOk+k=
+github.com/fsnotify/fsnotify v1.5.4 h1:jRbGcIw6P2Meqdwuo0H1p6JVLbL5DHKAKlYndzMwVZI=
+github.com/fsnotify/fsnotify v1.5.4/go.mod h1:OVB6XrOHzAwXMpEM7uPOzcehqUV2UqJxmVXmkdnm1bU=
 github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
 github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=

agent/utils/loader/load.go

@@ -18,6 +18,8 @@ import (
 	"fmt"
 	"os"
 
+	"github.com/aws/amazon-ecs-agent/agent/logger/field"
+
 	"github.com/aws/amazon-ecs-agent/agent/config"
 	"github.com/aws/amazon-ecs-agent/agent/dockerclient"
 	"github.com/aws/amazon-ecs-agent/agent/dockerclient/dockerapi"
@@ -35,7 +37,7 @@ type Loader interface {
 // GetContainerImage This function uses the DockerClient to inspect the image with the given name and tag.
 func GetContainerImage(imageName string, dockerClient dockerapi.DockerClient) (*types.ImageInspect, error) {
 	logger.Debug("Inspecting container image: ", logger.Fields{
-		imageName: imageName,
+		field.Image: imageName,
 	})
 
 	image, err := dockerClient.InspectImage(imageName)