Merge branch 'v3' into add-token-file-watcher

Author: sean-breen

api/grpc/mpi/v1/command.pb.go

@@ -90,6 +90,8 @@ message ContainerInfo {
     string container_id = 1 [(buf.validate.field).string.uuid = true];
     // The name of the host
     string hostname = 2 [(buf.validate.field).string.address = true];
+    // Release information of the container
+    ReleaseInfo release_info = 3;
 }
 
 // A response to a CreateConnectionRequest

api/grpc/mpi/v1/command.pb.validate.go

@@ -721,6 +721,7 @@ Container information
 | ----- | ---- | ----- | ----------- |
 | container_id | [string](#string) |  | The identifier of the container |
 | hostname | [string](#string) |  | The name of the host |
+| release_info | [ReleaseInfo](#mpi-v1-ReleaseInfo) |  | Release information of the container |
 
 
 

api/grpc/mpi/v1/command.proto

@@ -144,6 +144,7 @@ func (i *Info) ContainerInfo(ctx context.Context) *v1.Resource_ContainerInfo {
 		ContainerInfo: &v1.ContainerInfo{
 			ContainerId: i.getContainerID(),
 			Hostname:    hostname,
+			ReleaseInfo: i.getReleaseInfo(ctx, i.osReleaseLocation),
 		},
 	}
 }

docs/proto/protos.md

@@ -426,6 +426,20 @@ func TestInfo_IsContainer(t *testing.T) {
 
 func TestInfo_ContainerInfo(t *testing.T) {
 	ctx := context.Background()
+
+	osReleaseFile := helpers.CreateFileWithErrorCheck(t, os.TempDir(), "os-release")
+	defer helpers.RemoveFileWithErrorCheck(t, osReleaseFile.Name())
+	err := os.WriteFile(osReleaseFile.Name(), []byte(ubuntuReleaseInfo), os.ModeAppend)
+	require.NoError(t, err)
+
+	releaseInfo := &v1.ReleaseInfo{
+		Codename:  "jammy",
+		Id:        "ubuntu",
+		Name:      "Ubuntu",
+		VersionId: "22.04",
+		Version:   "22.04.5 LTS (Jammy Jellyfish)",
+	}
+
 	tests := []struct {
 		name              string
 		mountInfo         string
@@ -501,8 +515,9 @@ func TestInfo_ContainerInfo(t *testing.T) {
 
 			execMock := &execfakes.FakeExecInterface{}
 			execMock.HostnameReturns(test.expectHostname, nil)
+			execMock.ReleaseInfoReturns(releaseInfo)
 
-			_, err := mountInfoFile.WriteString(test.mountInfo)
+			_, err = mountInfoFile.WriteString(test.mountInfo)
 			require.NoError(tt, err)
 
 			err = mountInfoFile.Close()
@@ -515,6 +530,7 @@ func TestInfo_ContainerInfo(t *testing.T) {
 
 			assert.Equal(tt, test.expectContainerID, containerInfo.ContainerInfo.GetContainerId())
 			assert.Equal(tt, test.expectHostname, containerInfo.ContainerInfo.GetHostname())
+			assert.Equal(t, releaseInfo, containerInfo.ContainerInfo.GetReleaseInfo())
 		})
 	}
 }

internal/datasource/host/info.go

@@ -94,7 +94,7 @@ func (r *Resource) Process(ctx context.Context, msg *bus.Message) {
 
 			return
 		}
-		resource := r.resourceService.UpdateInstances(instanceList)
+		resource := r.resourceService.UpdateInstances(ctx, instanceList)
 
 		r.messagePipe.Process(ctx, &bus.Message{Topic: bus.ResourceUpdateTopic, Data: resource})
 
@@ -107,7 +107,7 @@ func (r *Resource) Process(ctx context.Context, msg *bus.Message) {
 
 			return
 		}
-		resource := r.resourceService.DeleteInstances(instanceList)
+		resource := r.resourceService.DeleteInstances(ctx, instanceList)
 
 		r.messagePipe.Process(ctx, &bus.Message{Topic: bus.ResourceUpdateTopic, Data: resource})
 

internal/datasource/host/info_test.go

@@ -16,6 +16,8 @@ import (
 	"strings"
 	"sync"
 
+	"google.golang.org/protobuf/proto"
+
 	"github.com/nginxinc/nginx-plus-go-client/v2/client"
 	"google.golang.org/protobuf/types/known/structpb"
 
@@ -42,8 +44,8 @@ const (
 
 type resourceServiceInterface interface {
 	AddInstances(instanceList []*mpi.Instance) *mpi.Resource
-	UpdateInstances(instanceList []*mpi.Instance) *mpi.Resource
-	DeleteInstances(instanceList []*mpi.Instance) *mpi.Resource
+	UpdateInstances(ctx context.Context, instanceList []*mpi.Instance) *mpi.Resource
+	DeleteInstances(ctx context.Context, instanceList []*mpi.Instance) *mpi.Resource
 	ApplyConfig(ctx context.Context, instanceID string) error
 	Instance(instanceID string) *mpi.Instance
 	GetHTTPUpstreamServers(ctx context.Context, instance *mpi.Instance, upstreams string) ([]client.UpstreamServer,
@@ -126,32 +128,45 @@ func (r *ResourceService) RemoveOperator(instanceList []*mpi.Instance) {
 	}
 }
 
-func (r *ResourceService) UpdateInstances(instanceList []*mpi.Instance) *mpi.Resource {
+func (r *ResourceService) UpdateInstances(ctx context.Context, instanceList []*mpi.Instance) *mpi.Resource {
 	r.resourceMutex.Lock()
 	defer r.resourceMutex.Unlock()
 
 	for _, updatedInstance := range instanceList {
-		for _, instance := range r.resource.GetInstances() {
-			if updatedInstance.GetInstanceMeta().GetInstanceId() == instance.GetInstanceMeta().GetInstanceId() {
-				instance.InstanceMeta = updatedInstance.GetInstanceMeta()
-				instance.InstanceRuntime = updatedInstance.GetInstanceRuntime()
-				instance.InstanceConfig = updatedInstance.GetInstanceConfig()
+		resourceCopy, ok := proto.Clone(r.resource).(*mpi.Resource)
+		if ok {
+			for _, instance := range resourceCopy.GetInstances() {
+				if updatedInstance.GetInstanceMeta().GetInstanceId() == instance.GetInstanceMeta().GetInstanceId() {
+					instance.InstanceMeta = updatedInstance.GetInstanceMeta()
+					instance.InstanceRuntime = updatedInstance.GetInstanceRuntime()
+					instance.InstanceConfig = updatedInstance.GetInstanceConfig()
+				}
 			}
+			r.resource = resourceCopy
+		} else {
+			slog.WarnContext(ctx, "Unable to clone resource while updating instances", "resource",
+				r.resource, "instances", instanceList)
 		}
 	}
 
 	return r.resource
 }
 
-func (r *ResourceService) DeleteInstances(instanceList []*mpi.Instance) *mpi.Resource {
+func (r *ResourceService) DeleteInstances(ctx context.Context, instanceList []*mpi.Instance) *mpi.Resource {
 	r.resourceMutex.Lock()
 	defer r.resourceMutex.Unlock()
 
 	for _, deletedInstance := range instanceList {
-		for index, instance := range r.resource.GetInstances() {
-			if deletedInstance.GetInstanceMeta().GetInstanceId() == instance.GetInstanceMeta().GetInstanceId() {
-				r.resource.Instances = append(r.resource.Instances[:index], r.resource.GetInstances()[index+1:]...)
+		resourceCopy, ok := proto.Clone(r.resource).(*mpi.Resource)
+		if ok {
+			for index, instance := range resourceCopy.GetInstances() {
+				if deletedInstance.GetInstanceMeta().GetInstanceId() == instance.GetInstanceMeta().GetInstanceId() {
+					r.resource.Instances = append(r.resource.Instances[:index], r.resource.GetInstances()[index+1:]...)
+				}
 			}
+		} else {
+			slog.WarnContext(ctx, "Unable to clone resource while deleting instances", "resource",
+				r.resource, "instances", instanceList)
 		}
 	}
 	r.RemoveOperator(instanceList)

internal/resource/resource_plugin.go

@@ -104,7 +104,7 @@ func TestResourceService_UpdateInstance(t *testing.T) {
 		t.Run(test.name, func(tt *testing.T) {
 			resourceService := NewResourceService(ctx, types.AgentConfig())
 			resourceService.resource.Instances = []*v1.Instance{protos.GetNginxOssInstance([]string{})}
-			resource := resourceService.UpdateInstances(test.instanceList)
+			resource := resourceService.UpdateInstances(ctx, test.instanceList)
 			assert.Equal(tt, test.resource.GetInstances(), resource.GetInstances())
 		})
 	}
@@ -141,7 +141,7 @@ func TestResourceService_DeleteInstance(t *testing.T) {
 				protos.GetNginxOssInstance([]string{}),
 				protos.GetNginxPlusInstance([]string{}),
 			}
-			resource := resourceService.DeleteInstances(test.instanceList)
+			resource := resourceService.DeleteInstances(ctx, test.instanceList)
 			assert.Equal(tt, test.resource.GetInstances(), resource.GetInstances())
 		})
 	}