deployment status (#262)

Deployer forwards deployment status of each agents to controller.
It deployes/revokes each agent one after one and pushes the status of operation.

api/computes_components.partials.yml

@@ -92,26 +92,10 @@ AgentKVs:
   additionalProperties:
     type: string
 
-########################################
-# Deployment status within a compute cluster
-########################################
-DeploymentStatus:
-  description: Deployment status within a compute cluster
-  type: object
-  properties:
-    jobId:
-      type: string
-    agentStatuses:
-      type: array
-      items: 
-        $ref: '#/components/schemas/AgentStatus'
-  required:
-    - jobId
-
 ########################################
 # Status for an agent within a deployment
 ########################################
-AgentStatus:
+DeploymentStatus:
   description: Status for an agent within a deployment
   type: object
   additionalProperties:
@@ -122,6 +106,7 @@ AgentStatus:
 ########################################
 AgentState:
   enum:
-    - agentDeployed
-    - agentRevoked
-    - agentFailed
+    - agentDeploySuccess
+    - agentDeployFailed
+    - agentRevokeSuccess
+    - agentRevokeFailed

cmd/controller/app/database/db_interfaces.go

@@ -91,4 +91,6 @@ type ComputeService interface {
 	RegisterCompute(openapi.ComputeSpec) (openapi.ComputeStatus, error)
 	GetComputeIdsByRegion(string) ([]string, error)
 	GetComputeById(string) (openapi.ComputeSpec, error)
+	// UpdateDeploymentStatus call replaces existing agent statuses with received statuses in collection.
+	UpdateDeploymentStatus(computeId string, jobId string, agentStatuses map[string]openapi.AgentState) error
 }

cmd/controller/app/database/mongodb/compute.go

@@ -173,3 +173,32 @@ func (db *MongoService) GetComputeById(computeId string) (openapi.ComputeSpec, e
 	}
 	return currentDocument, nil
 }
+
+func (db *MongoService) UpdateDeploymentStatus(computeId string, jobId string, agentStatuses map[string]openapi.AgentState) error {
+	filter := bson.M{util.DBFieldJobId: jobId, util.DBFieldComputeId: computeId}
+	result := db.deploymentCollection.FindOne(context.TODO(), filter)
+	// update fields if document is already present
+	if result.Err() == nil {
+		result = db.deploymentCollection.FindOneAndUpdate(context.TODO(),
+			filter,
+			bson.M{"$set": bson.M{util.DBFieldAgentStatuses: agentStatuses}})
+		if err := result.Err(); err != nil {
+			err = ErrorCheck(err)
+			zap.S().Errorf("Failed to update new deployment status: %v", err)
+			return err
+		}
+	} else {
+		_, err := db.deploymentCollection.InsertOne(context.TODO(),
+			bson.M{util.DBFieldJobId: jobId,
+				util.DBFieldComputeId:     computeId,
+				util.DBFieldAgentStatuses: agentStatuses,
+			})
+		if err != nil {
+			err = ErrorCheck(err)
+			zap.S().Errorf("Failed to create new deployment status: %v", err)
+			return err
+		}
+	}
+	zap.S().Debugf("New deployment status for jobid: %s inserted", jobId)
+	return nil
+}

cmd/controller/app/database/mongodb/compute_test.go

@@ -0,0 +1,37 @@
+package mongodb
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo/integration/mtest"
+
+	"github.com/cisco-open/flame/pkg/openapi"
+)
+
+func TestMongoService_UpdateDeploymentStatus(t *testing.T) {
+	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
+	defer mt.Close()
+	mt.Run("success", func(mt *mtest.T) {
+		db := &MongoService{
+			deploymentCollection: mt.Coll,
+		}
+		mt.AddMockResponses(mtest.CreateSuccessResponse(), mtest.CreateCursorResponse(1, "flame.deployment", mtest.FirstBatch, bson.D{}))
+		err := db.UpdateDeploymentStatus("test jobid","test compute id",
+			map[string]openapi.AgentState{"test task id": openapi.AGENT_DEPLOY_SUCCESS,
+		})
+		assert.Nil(t, err)
+	})
+	mt.Run("status update failure", func(mt *mtest.T) {
+		db := &MongoService{
+			deploymentCollection: mt.Coll,
+		}
+		mt.AddMockResponses(
+			mtest.CreateCursorResponse(1, "flame.deployment", mtest.FirstBatch, bson.D{{"_id", "1"}}))
+		err := db.UpdateDeploymentStatus("test jobid", "test compute id",
+			map[string]openapi.AgentState{"test task id": openapi.AGENT_DEPLOY_SUCCESS},
+		)
+		assert.NotNil(t, err)
+	})
+}

cmd/controller/app/database/mongodb/service.go

@@ -25,16 +25,19 @@ import (
 	"go.mongodb.org/mongo-driver/mongo/readpref"
 	"go.mongodb.org/mongo-driver/x/bsonx"
 	"go.uber.org/zap"
+
+	"github.com/cisco-open/flame/pkg/util"
 )
 
 const (
 	DatabaseName = "flame"
 
-	ComputeCollection = "t_compute"
-	DatasetCollection = "t_dataset"
-	DesignCollection  = "t_design"
-	JobCollection     = "t_job"
-	TaskCollection    = "t_task"
+	ComputeCollection    = "t_compute"
+	DatasetCollection    = "t_dataset"
+	DesignCollection     = "t_design"
+	JobCollection        = "t_job"
+	TaskCollection       = "t_task"
+	DeploymentCollection = "t_deployment"
 
 	orderAscend  int32 = 1
 	orderDescend int32 = -1
@@ -44,11 +47,12 @@ type MongoService struct {
 	client   *mongo.Client
 	database *mongo.Database
 
-	computeCollection *mongo.Collection
-	datasetCollection *mongo.Collection
-	designCollection  *mongo.Collection
-	jobCollection     *mongo.Collection
-	taskCollection    *mongo.Collection
+	computeCollection    *mongo.Collection
+	datasetCollection    *mongo.Collection
+	designCollection     *mongo.Collection
+	jobCollection        *mongo.Collection
+	taskCollection       *mongo.Collection
+	deploymentCollection *mongo.Collection
 }
 
 type uniqueIndexInfo struct {
@@ -91,31 +95,36 @@ func NewMongoService(uri string) (*MongoService, error) {
 
 	db := client.Database(DatabaseName)
 	mongoDB := &MongoService{
-		client:            client,
-		database:          db,
-		computeCollection: db.Collection(ComputeCollection),
-		datasetCollection: db.Collection(DatasetCollection),
-		designCollection:  db.Collection(DesignCollection),
-		jobCollection:     db.Collection(JobCollection),
-		taskCollection:    db.Collection(TaskCollection),
+		client:               client,
+		database:             db,
+		computeCollection:    db.Collection(ComputeCollection),
+		datasetCollection:    db.Collection(DatasetCollection),
+		designCollection:     db.Collection(DesignCollection),
+		jobCollection:        db.Collection(JobCollection),
+		taskCollection:       db.Collection(TaskCollection),
+		deploymentCollection: db.Collection(DeploymentCollection),
 	}
 
 	uiiList := []uniqueIndexInfo{
 		{
 			mongoDB.computeCollection,
-			map[string]int32{"computeid": orderAscend, "region": orderAscend},
+			map[string]int32{util.DBFieldComputeId: orderAscend, util.DBFieldComputeRegion: orderAscend},
 		},
 		{
 			mongoDB.datasetCollection,
-			map[string]int32{"userid": orderAscend, "url": orderAscend},
+			map[string]int32{util.DBFieldUserId: orderAscend, util.DBFieldURL: orderAscend},
 		},
 		{
 			mongoDB.designCollection,
-			map[string]int32{"userid": orderAscend, "id": orderAscend},
+			map[string]int32{util.DBFieldUserId: orderAscend, util.DBFieldId: orderAscend},
 		},
 		{
 			mongoDB.taskCollection,
-			map[string]int32{"jobid": orderAscend, "taskid": orderAscend},
+			map[string]int32{util.DBFieldJobId: orderAscend, util.DBFieldTaskId: orderAscend},
+		},
+		{
+			mongoDB.deploymentCollection,
+			map[string]int32{util.DBFieldComputeId: orderAscend, util.DBFieldJobId: orderAscend},
 		},
 	}
 	for _, uii := range uiiList {

cmd/deployer/app/deployer/k8s.go

@@ -86,6 +86,9 @@ func (deployer *K8sDeployer) Install(releaseName string, chartPath string) error
 	installObj.ReleaseName = releaseName
 	release, err := installObj.Run(chart, nil)
 	if err != nil {
+		if release != nil && release.Info != nil {
+			zap.S().Errorf("Release %s failed with status : %v", releaseName, release.Info.Status)
+		}
 		return err
 	}