feat(scheduler): Implement POST /intervalaction V2 API (#3190)

* feat(scheduler): Implement POST /intervalaction V2 API

Implement the V2 API according to the swagger doc https://app.swaggerhub.com/apis-docs/EdgeXFoundry1/support-scheduler/2.x#/default/post_intervalaction

Close #3187

* feat(scheduler): Use common Address for IntervalAction API

* feat(scheduler): Upgrade core-contract lib and update swagger file

- Upgrade core-contract lib to latest version
- Add the description to clarify that id or name is an either requirement, not both

Signed-off-by: weichou <weichou1229@gmail.com>

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/edgexfoundry/go-mod-bootstrap/v2 v2.0.0-dev.13
 	github.com/edgexfoundry/go-mod-configuration/v2 v2.0.0-dev.3
-	github.com/edgexfoundry/go-mod-core-contracts/v2 v2.0.0-dev.43
+	github.com/edgexfoundry/go-mod-core-contracts/v2 v2.0.0-dev.46
 	github.com/edgexfoundry/go-mod-messaging/v2 v2.0.0-dev.6
 	github.com/edgexfoundry/go-mod-registry/v2 v2.0.0-dev.3
 	github.com/edgexfoundry/go-mod-secrets/v2 v2.0.0-dev.7

internal/pkg/v2/infrastructure/redis/client.go

@@ -742,6 +742,17 @@ func (c *Client) DeleteIntervalByName(name string) errors.EdgeX {
 	return nil
 }
 
+// AddIntervalAction adds a new intervalAction
+func (c *Client) AddIntervalAction(action model.IntervalAction) (model.IntervalAction, errors.EdgeX) {
+	conn := c.Pool.Get()
+	defer conn.Close()
+
+	if len(action.Id) == 0 {
+		action.Id = uuid.New().String()
+	}
+	return addIntervalAction(conn, action)
+}
+
 // AddSubscription adds a new subscription
 func (c *Client) AddSubscription(subscription model.Subscription) (model.Subscription, errors.EdgeX) {
 	conn := c.Pool.Get()

internal/pkg/v2/infrastructure/redis/intervalaction.go

@@ -0,0 +1,70 @@
+//
+// Copyright (C) 2021 IOTech Ltd
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package redis
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/edgexfoundry/edgex-go/internal/pkg/v2/utils"
+
+	"github.com/edgexfoundry/go-mod-core-contracts/v2/errors"
+	"github.com/edgexfoundry/go-mod-core-contracts/v2/v2"
+	"github.com/edgexfoundry/go-mod-core-contracts/v2/v2/models"
+
+	"github.com/gomodule/redigo/redis"
+)
+
+const (
+	IntervalActionCollection       = "ss|ia"
+	IntervalActionCollectionName   = IntervalActionCollection + DBKeySeparator + v2.Name
+	IntervalActionCollectionTarget = IntervalActionCollection + DBKeySeparator + v2.Target
+)
+
+// intervalActionStoredKey return the intervalAction's stored key which combines the collection name and object id
+func intervalActionStoredKey(id string) string {
+	return CreateKey(IntervalActionCollection, id)
+}
+
+// addIntervalAction adds a new intervalAction into DB
+func addIntervalAction(conn redis.Conn, action models.IntervalAction) (models.IntervalAction, errors.EdgeX) {
+	exists, edgeXerr := objectIdExists(conn, intervalActionStoredKey(action.Id))
+	if edgeXerr != nil {
+		return action, errors.NewCommonEdgeXWrapper(edgeXerr)
+	} else if exists {
+		return action, errors.NewCommonEdgeX(errors.KindDuplicateName, fmt.Sprintf("intervalAction id %s already exists", action.Id), edgeXerr)
+	}
+
+	exists, edgeXerr = objectNameExists(conn, IntervalActionCollectionName, action.Name)
+	if edgeXerr != nil {
+		return action, errors.NewCommonEdgeXWrapper(edgeXerr)
+	} else if exists {
+		return action, errors.NewCommonEdgeX(errors.KindDuplicateName, fmt.Sprintf("intervalAction name %s already exists", action.Name), edgeXerr)
+	}
+
+	ts := utils.MakeTimestamp()
+	if action.Created == 0 {
+		action.Created = ts
+	}
+	action.Modified = ts
+
+	m, err := json.Marshal(action)
+	if err != nil {
+		return action, errors.NewCommonEdgeX(errors.KindContractInvalid, "unable to JSON marshal intervalAction for Redis persistence", err)
+	}
+
+	storedKey := intervalActionStoredKey(action.Id)
+	_ = conn.Send(MULTI)
+	_ = conn.Send(SET, storedKey, m)
+	_ = conn.Send(ZADD, IntervalActionCollection, action.Modified, storedKey)
+	_ = conn.Send(HSET, IntervalActionCollectionName, action.Name, storedKey)
+	_, err = conn.Do(EXEC)
+	if err != nil {
+		edgeXerr = errors.NewCommonEdgeX(errors.KindDatabaseError, "intervalAction creation failed", err)
+	}
+
+	return action, edgeXerr
+}

internal/support/scheduler/v2/application/interval.go

@@ -22,11 +22,11 @@ import (
 
 // The AddInterval function accepts the new Interval model from the controller function
 // and then invokes AddInterval function of infrastructure layer to add new Interval
-func AddInterval(d models.Interval, ctx context.Context, dic *di.Container) (id string, edgeXerr errors.EdgeX) {
+func AddInterval(interval models.Interval, ctx context.Context, dic *di.Container) (id string, edgeXerr errors.EdgeX) {
 	dbClient := v2SchedulerContainer.DBClientFrom(dic.Get)
 	lc := container.LoggingClientFrom(dic.Get)
 
-	addedInterval, err := dbClient.AddInterval(d)
+	addedInterval, err := dbClient.AddInterval(interval)
 	if err != nil {
 		return "", errors.NewCommonEdgeXWrapper(err)
 	}

internal/support/scheduler/v2/application/intervalaction.go

@@ -0,0 +1,42 @@
+//
+// Copyright (C) 2021 IOTech Ltd
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package application
+
+import (
+	"context"
+
+	"github.com/edgexfoundry/edgex-go/internal/pkg/correlation"
+	v2SchedulerContainer "github.com/edgexfoundry/edgex-go/internal/support/scheduler/v2/bootstrap/container"
+
+	"github.com/edgexfoundry/go-mod-bootstrap/v2/bootstrap/container"
+	"github.com/edgexfoundry/go-mod-bootstrap/v2/di"
+	"github.com/edgexfoundry/go-mod-core-contracts/v2/errors"
+	"github.com/edgexfoundry/go-mod-core-contracts/v2/v2/models"
+)
+
+// The AddIntervalAction function accepts the new IntervalAction model from the controller function
+// and then invokes AddIntervalAction function of infrastructure layer to add new IntervalAction
+func AddIntervalAction(action models.IntervalAction, ctx context.Context, dic *di.Container) (id string, edgeXerr errors.EdgeX) {
+	dbClient := v2SchedulerContainer.DBClientFrom(dic.Get)
+	lc := container.LoggingClientFrom(dic.Get)
+
+	// checks the interval existence by name
+	_, edgeXerr = dbClient.IntervalByName(action.IntervalName)
+	if edgeXerr != nil {
+		return id, errors.NewCommonEdgeXWrapper(edgeXerr)
+	}
+
+	addedAction, err := dbClient.AddIntervalAction(action)
+	if err != nil {
+		return "", errors.NewCommonEdgeXWrapper(err)
+	}
+
+	lc.Debugf("IntervalAction created on DB successfully. IntervalAction ID: %s, Correlation-ID: %s ",
+		addedAction.Id,
+		correlation.FromContext(ctx))
+
+	return addedAction.Id, nil
+}

internal/support/scheduler/v2/controller/http/const_test.go

@@ -12,4 +12,9 @@ const (
 	TestIntervalEnd       = "20190802T150405"
 	TestIntervalFrequency = "30ms"
 	TestIntervalRunOnce   = false
+
+	TestIntervalActionName = "TestIntervalAction"
+	TestHost               = "localhost"
+	TestPort               = 48089
+	TestHTTPMethod         = "GET"
 )

internal/support/scheduler/v2/controller/http/intervalaction.go

@@ -0,0 +1,88 @@
+//
+// Copyright (C) 2021 IOTech Ltd
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package http
+
+import (
+	"net/http"
+
+	"github.com/edgexfoundry/edgex-go/internal/pkg"
+	"github.com/edgexfoundry/edgex-go/internal/pkg/correlation"
+	"github.com/edgexfoundry/edgex-go/internal/pkg/v2/utils"
+	"github.com/edgexfoundry/edgex-go/internal/support/scheduler/v2/application"
+	"github.com/edgexfoundry/edgex-go/internal/support/scheduler/v2/io"
+
+	"github.com/edgexfoundry/go-mod-bootstrap/v2/bootstrap/container"
+	"github.com/edgexfoundry/go-mod-bootstrap/v2/di"
+	"github.com/edgexfoundry/go-mod-core-contracts/v2/clients"
+	commonDTO "github.com/edgexfoundry/go-mod-core-contracts/v2/v2/dtos/common"
+	requestDTO "github.com/edgexfoundry/go-mod-core-contracts/v2/v2/dtos/requests"
+)
+
+type IntervalActionController struct {
+	reader io.IntervalActionReader
+	dic    *di.Container
+}
+
+// NewIntervalActionController creates and initializes an IntervalActionController
+func NewIntervalActionController(dic *di.Container) *IntervalActionController {
+	return &IntervalActionController{
+		reader: io.NewIntervalActionRequestReader(),
+		dic:    dic,
+	}
+}
+
+func (dc *IntervalActionController) AddIntervalAction(w http.ResponseWriter, r *http.Request) {
+	if r.Body != nil {
+		defer func() { _ = r.Body.Close() }()
+	}
+
+	lc := container.LoggingClientFrom(dc.dic.Get)
+
+	ctx := r.Context()
+	correlationId := correlation.FromContext(ctx)
+
+	actionDTOs, err := dc.reader.ReadAddIntervalActionRequest(r.Body)
+	if err != nil {
+		lc.Error(err.Error(), clients.CorrelationHeader, correlationId)
+		lc.Debug(err.DebugMessages(), clients.CorrelationHeader, correlationId)
+		errResponses := commonDTO.NewBaseResponse(
+			"",
+			err.Message(),
+			err.Code())
+		utils.WriteHttpHeader(w, ctx, err.Code())
+		pkg.Encode(errResponses, w, lc)
+		return
+	}
+	actions := requestDTO.AddIntervalActionReqToIntervalActionModels(actionDTOs)
+
+	var addResponses []interface{}
+	for i, action := range actions {
+		var response interface{}
+		reqId := actionDTOs[i].RequestId
+		newId, err := application.AddIntervalAction(action, ctx, dc.dic)
+		if err != nil {
+			lc.Error(err.Error(), clients.CorrelationHeader, correlationId)
+			lc.Debug(err.DebugMessages(), clients.CorrelationHeader, correlationId)
+			response = commonDTO.NewBaseResponse(
+				reqId,
+				err.Message(),
+				err.Code())
+		} else {
+			response = commonDTO.NewBaseWithIdResponse(
+				reqId,
+				"",
+				http.StatusCreated,
+				newId)
+		}
+		addResponses = append(addResponses, response)
+
+		// TODO Add the new IntervalAction into scheduler queue
+		//err = scClient.AddIntervalActionToQueue(intervalAction)
+	}
+
+	utils.WriteHttpHeader(w, ctx, http.StatusMultiStatus)
+	pkg.Encode(addResponses, w, lc)
+}

internal/support/scheduler/v2/controller/http/intervalaction_test.go

@@ -0,0 +1,108 @@
+//
+// Copyright (C) 2021 IOTech Ltd
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package http
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	v2SchedulerContainer "github.com/edgexfoundry/edgex-go/internal/support/scheduler/v2/bootstrap/container"
+	dbMock "github.com/edgexfoundry/edgex-go/internal/support/scheduler/v2/infrastructure/interfaces/mocks"
+
+	"github.com/edgexfoundry/go-mod-bootstrap/v2/di"
+	"github.com/edgexfoundry/go-mod-core-contracts/v2/errors"
+	"github.com/edgexfoundry/go-mod-core-contracts/v2/v2"
+	"github.com/edgexfoundry/go-mod-core-contracts/v2/v2/dtos"
+	"github.com/edgexfoundry/go-mod-core-contracts/v2/v2/dtos/common"
+	"github.com/edgexfoundry/go-mod-core-contracts/v2/v2/dtos/requests"
+	"github.com/edgexfoundry/go-mod-core-contracts/v2/v2/models"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func addIntervalActionRequestData() requests.AddIntervalActionRequest {
+	restAddress := dtos.NewRESTAddress(TestHost, TestPort, TestHTTPMethod)
+	intervalAction := dtos.NewIntervalAction(TestIntervalActionName, TestIntervalName, restAddress)
+	return requests.NewAddIntervalActionRequest(intervalAction)
+}
+
+func TestAddIntervalAction(t *testing.T) {
+	dic := mockDic()
+	dbClientMock := &dbMock.DBClient{}
+
+	valid := addIntervalActionRequestData()
+	model := dtos.ToIntervalActionModel(valid.Action)
+	dbClientMock.On("IntervalByName", model.IntervalName).Return(models.Interval{}, nil)
+	dbClientMock.On("AddIntervalAction", model).Return(model, nil)
+
+	noName := valid
+	noName.Action.Name = ""
+	noRequestId := valid
+	noRequestId.RequestId = ""
+
+	duplicatedName := valid
+	duplicatedName.Action.Name = "duplicatedName"
+	model = dtos.ToIntervalActionModel(duplicatedName.Action)
+	dbClientMock.On("AddIntervalAction", model).Return(model, errors.NewCommonEdgeX(errors.KindDuplicateName, fmt.Sprintf("intervalAction name %s already exists", model.Name), nil))
+
+	dic.Update(di.ServiceConstructorMap{
+		v2SchedulerContainer.DBClientInterfaceName: func(get di.Get) interface{} {
+			return dbClientMock
+		},
+	})
+	controller := NewIntervalActionController(dic)
+	require.NotNil(t, controller)
+	tests := []struct {
+		name               string
+		request            []requests.AddIntervalActionRequest
+		expectedStatusCode int
+	}{
+		{"Valid", []requests.AddIntervalActionRequest{valid}, http.StatusCreated},
+		{"Valid - no request Id", []requests.AddIntervalActionRequest{noRequestId}, http.StatusCreated},
+		{"Invalid - no name", []requests.AddIntervalActionRequest{noName}, http.StatusBadRequest},
+		{"Invalid - duplicated name", []requests.AddIntervalActionRequest{duplicatedName}, http.StatusConflict},
+	}
+	for _, testCase := range tests {
+		t.Run(testCase.name, func(t *testing.T) {
+			jsonData, err := json.Marshal(testCase.request)
+			require.NoError(t, err)
+
+			reader := strings.NewReader(string(jsonData))
+			req, err := http.NewRequest(http.MethodPost, v2.ApiIntervalActionRoute, reader)
+			require.NoError(t, err)
+
+			// Act
+			recorder := httptest.NewRecorder()
+			handler := http.HandlerFunc(controller.AddIntervalAction)
+			handler.ServeHTTP(recorder, req)
+			if testCase.expectedStatusCode == http.StatusBadRequest {
+				var res common.BaseResponse
+				err = json.Unmarshal(recorder.Body.Bytes(), &res)
+				require.NoError(t, err)
+
+				// Assert
+				assert.Equal(t, testCase.expectedStatusCode, recorder.Result().StatusCode, "HTTP status code not as expected")
+				assert.Equal(t, v2.ApiVersion, res.ApiVersion, "API Version not as expected")
+				assert.Equal(t, testCase.expectedStatusCode, res.StatusCode, "BaseResponse status code not as expected")
+				assert.NotEmpty(t, res.Message, "Message is empty")
+			} else {
+				var res []common.BaseResponse
+				err = json.Unmarshal(recorder.Body.Bytes(), &res)
+				require.NoError(t, err)
+
+				// Assert
+				assert.Equal(t, http.StatusMultiStatus, recorder.Result().StatusCode, "HTTP status code not as expected")
+				assert.Equal(t, v2.ApiVersion, res[0].ApiVersion, "API Version not as expected")
+				assert.Equal(t, testCase.expectedStatusCode, res[0].StatusCode, "BaseResponse status code not as expected")
+			}
+		})
+	}
+}

internal/support/scheduler/v2/infrastructure/interfaces/db.go

@@ -19,4 +19,6 @@ type DBClient interface {
 	AllIntervals(offset int, limit int) ([]model.Interval, errors.EdgeX)
 	DeleteIntervalByName(name string) errors.EdgeX
 	UpdateInterval(interval model.Interval) errors.EdgeX
+
+	AddIntervalAction(e model.IntervalAction) (model.IntervalAction, errors.EdgeX)
 }