feat(data): Implement DELETE /event/device/name/{name} V2 API (#2874)

* feat(data): Implement DELETE /event/device/name/{name} V2 API

Implement DELETE /event/device/name/{name} for V2 API.  The current V2 API doc
still lists DELETE /event/device/{deviceId}; however, as https://github.com/edgexfoundry/go-mod-core-contracts/blob/master/v2/models/event.go#L14
has been updated to DeviceName instead of DeviceId, the V2 API spec shall be
updated to reflect this change.
Update core-data.yaml with removal of 404 response
Fix #2837


Signed-off-by: Jude Hung <[email protected]>

internal/core/data/v2/application/event.go

@@ -9,6 +9,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"strings"
 
 	dataContainer "github.com/edgexfoundry/edgex-go/internal/core/data/container"
 	v2DataContainer "github.com/edgexfoundry/edgex-go/internal/core/data/v2/bootstrap/container"
@@ -170,6 +171,22 @@ func DeletePushedEvents(dic *di.Container) errors.EdgeX {
 	return nil
 }
 
+// The DeleteEventsByDeviceName function will be invoked by controller functions
+// and then invokes DeleteEventsByDeviceName function in the infrastructure layer to remove
+// all events/readings that are associated with the given deviceName
+func DeleteEventsByDeviceName(deviceName string, dic *di.Container) errors.EdgeX {
+	if len(strings.TrimSpace(deviceName)) <= 0 {
+		return errors.NewCommonEdgeX(errors.KindInvalidId, "blank device name is not allowed", nil)
+	}
+	dbClient := v2DataContainer.DBClientFrom(dic.Get)
+
+	err := dbClient.DeleteEventsByDeviceName(deviceName)
+	if err != nil {
+		return errors.NewCommonEdgeXWrapper(err)
+	}
+	return nil
+}
+
 // UpdateEventPushedById updates event pushed timestamp per incoming event id
 func UpdateEventPushedById(id string, dic *di.Container) errors.EdgeX {
 	dbClient := v2DataContainer.DBClientFrom(dic.Get)

internal/core/data/v2/application/event_test.go

@@ -90,6 +90,7 @@ func newMockDB(persist bool) *dbMock.DBClient {
 		myMock.On("EventTotalCount").Return(testEventCount, nil)
 		myMock.On("EventCountByDevice", testDeviceName).Return(testEventCount, nil)
 		myMock.On("DeletePushedEvents").Return(nil)
+		myMock.On("DeleteEventsByDeviceName", testDeviceName).Return(nil)
 	}
 
 	return myMock
@@ -272,3 +273,41 @@ func TestDeletePushedEvents(t *testing.T) {
 	err := DeletePushedEvents(dic)
 	require.NoError(t, err)
 }
+
+func TestDeleteEventsByDeviceName(t *testing.T) {
+	dbClientMock := newMockDB(true)
+
+	dic := mocks.NewMockDIC()
+	dic.Update(di.ServiceConstructorMap{
+		v2DataContainer.DBClientInterfaceName: func(get di.Get) interface{} {
+			return dbClientMock
+		},
+	})
+
+	tests := []struct {
+		Name               string
+		deviceName         string
+		ErrorExpected      bool
+		ExpectedErrKind    errors.ErrKind
+		ExpectedStatusCode int
+	}{
+		{"Valid - Delete Event by Id", testDeviceName, false, errors.KindInvalidId, http.StatusOK},
+		{"Invalid - Empty device name", "", true, errors.KindInvalidId, http.StatusBadRequest},
+		{"Invalid - Empty device name with spaces", " \n\t\r ", true, errors.KindInvalidId, http.StatusBadRequest},
+	}
+
+	for _, testCase := range tests {
+		t.Run(testCase.Name, func(t *testing.T) {
+			err := DeleteEventsByDeviceName(testCase.deviceName, dic)
+
+			if testCase.ErrorExpected {
+				require.Error(t, err)
+				assert.NotEmpty(t, err.Error(), "Error message is empty")
+				assert.Equal(t, testCase.ExpectedErrKind, errors.Kind(err), "Error kind not as expected")
+				assert.Equal(t, testCase.ExpectedStatusCode, err.Code(), "Error code not as expected")
+			} else {
+				require.NoError(t, err)
+			}
+		})
+	}
+}

internal/core/data/v2/controller/http/event.go

@@ -352,3 +352,32 @@ func (ec *EventController) EventsByDeviceName(w http.ResponseWriter, r *http.Req
 	utils.WriteHttpHeader(w, ctx, statusCode)
 	pkg.Encode(response, w, lc)
 }
+
+func (ec *EventController) DeleteEventsByDeviceName(w http.ResponseWriter, r *http.Request) {
+	// retrieve all the service injections from bootstrap
+	lc := container.LoggingClientFrom(ec.dic.Get)
+
+	ctx := r.Context()
+	correlationId := correlation.FromContext(ctx)
+
+	vars := mux.Vars(r)
+	deviceName := vars[v2.Name]
+
+	var response interface{}
+	var statusCode int
+
+	// Delete events with associated Device deviceName
+	err := application.DeleteEventsByDeviceName(deviceName, ec.dic)
+	if err != nil {
+		lc.Debug(err.DebugMessages(), clients.CorrelationHeader, correlationId)
+		response = commonDTO.NewBaseResponse("", err.Message(), err.Code())
+		statusCode = err.Code()
+	} else {
+		response = commonDTO.NewBaseResponse("", "", http.StatusAccepted)
+		statusCode = http.StatusAccepted
+	}
+
+	utils.WriteHttpHeader(w, ctx, statusCode)
+	// encode and send out the response
+	pkg.Encode(response, w, lc)
+}

internal/core/data/v2/controller/http/event_test.go

@@ -437,6 +437,35 @@ func TestDeletePushedEvents(t *testing.T) {
 	assert.Empty(t, actualResponse.Message, "Message should be empty when it is successful")
 }
 
+func TestDeleteEventsByDeviceName(t *testing.T) {
+	deviceName := "deviceA"
+	dbClientMock := &dbMock.DBClient{}
+	dbClientMock.On("DeleteEventsByDeviceName", deviceName).Return(nil)
+
+	dic := mocks.NewMockDIC()
+	dic.Update(di.ServiceConstructorMap{
+		v2DataContainer.DBClientInterfaceName: func(get di.Get) interface{} {
+			return dbClientMock
+		},
+	})
+	ec := NewEventController(dic)
+
+	req, err := http.NewRequest(http.MethodDelete, v2.ApiEventByDeviceNameRoute, http.NoBody)
+	req = mux.SetURLVars(req, map[string]string{v2.Name: deviceName})
+	require.NoError(t, err)
+
+	recorder := httptest.NewRecorder()
+	handler := http.HandlerFunc(ec.DeleteEventsByDeviceName)
+	handler.ServeHTTP(recorder, req)
+
+	var actualResponse common.BaseResponse
+	err = json.Unmarshal(recorder.Body.Bytes(), &actualResponse)
+
+	assert.Equal(t, v2.ApiVersion, actualResponse.ApiVersion, "API Version not as expected")
+	assert.Equal(t, http.StatusAccepted, recorder.Result().StatusCode, "HTTP status code not as expected")
+	assert.Empty(t, actualResponse.Message, "Message should be empty when it is successful")
+}
+
 func TestUpdateEventPushedById(t *testing.T) {
 	expectedResponseCode := http.StatusMultiStatus
 

internal/core/data/v2/infrastructure/interfaces/db.go

@@ -22,4 +22,5 @@ type DBClient interface {
 	AllEvents(offset int, limit int) ([]model.Event, errors.EdgeX)
 	EventsByDeviceName(offset int, limit int, name string) ([]model.Event, errors.EdgeX)
 	DeletePushedEvents() errors.EdgeX
+	DeleteEventsByDeviceName(deviceName string) errors.EdgeX
 }

internal/core/data/v2/router.go

@@ -33,6 +33,7 @@ func LoadRestRoutes(r *mux.Router, dic *di.Container) {
 	r.HandleFunc(v2Constant.ApiAllEventRoute, ec.AllEvents).Methods(http.MethodGet)
 	r.HandleFunc(v2Constant.ApiEventByDeviceNameRoute, ec.EventsByDeviceName).Methods(http.MethodGet)
 	r.HandleFunc(v2Constant.ApiEventScrubRoute, ec.DeletePushedEvents).Methods(http.MethodDelete)
+	r.HandleFunc(v2Constant.ApiEventByDeviceNameRoute, ec.DeleteEventsByDeviceName).Methods(http.MethodDelete)
 
 	r.Use(correlation.ManageHeader)
 	r.Use(correlation.OnResponseComplete)

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

@@ -87,7 +87,25 @@ func (c *Client) DeletePushedEvents() (edgeXerr errors.EdgeX) {
 	conn := c.Pool.Get()
 	defer conn.Close()
 
-	eventIds, readingIds, err := getPushedEventReadingIds(conn)
+	eventIds, readingIds, err := getEventReadingIdsByKey(conn, EventsCollectionPushed)
+	if err != nil {
+		return errors.NewCommonEdgeXWrapper(err)
+	}
+	c.loggingClient.Debug(fmt.Sprintf("Prepare to delete %v readings", len(readingIds)))
+	go c.asyncDeleteReadingsByIds(readingIds)
+	c.loggingClient.Debug(fmt.Sprintf("Prepare to delete %v events", len(eventIds)))
+	go c.asyncDeleteEventsByIds(eventIds)
+
+	return nil
+}
+
+// DeleteEventsByDeviceName deletes all pushed events and corresponding readings.  This function is implemented to starts up
+// two goroutines to delete readings and events in the bckground to achieve better performance.
+func (c *Client) DeleteEventsByDeviceName(deviceName string) (edgeXerr errors.EdgeX) {
+	conn := c.Pool.Get()
+	defer conn.Close()
+
+	eventIds, readingIds, err := getEventReadingIdsByKey(conn, fmt.Sprintf("%s:%s", EventsCollectionDeviceName, deviceName))
 	if err != nil {
 		return errors.NewCommonEdgeXWrapper(err)
 	}
@@ -205,28 +223,28 @@ func deleteEventById(conn redis.Conn, id string) (edgeXerr errors.EdgeX) {
 	return edgeXerr
 }
 
-func getPushedEventReadingIds(conn redis.Conn) (eventIds []string, readingIds []string, edgeXerr errors.EdgeX) {
-	pushedEventIds, err := redis.Strings(conn.Do(ZRANGEBYSCORE, EventsCollectionPushed, GreaterThanZero, InfiniteMax))
+func getEventReadingIdsByKey(conn redis.Conn, key string) (eventIds []string, readingIds []string, edgeXerr errors.EdgeX) {
+	eventIds, err := redis.Strings(conn.Do(ZRANGEBYSCORE, key, GreaterThanZero, InfiniteMax))
 	if err != nil {
-		return nil, nil, errors.NewCommonEdgeX(errors.KindDatabaseError, "retrieve all pushed event ids failed", err)
+		return nil, nil, errors.NewCommonEdgeX(errors.KindDatabaseError, fmt.Sprintf("retrieve event ids by key %s failed", key), err)
 	}
-	pushedEvents, edgeXerr := getObjectsByIds(conn, common.ConvertStringsToInterfaces(pushedEventIds))
+	events, edgeXerr := getObjectsByIds(conn, common.ConvertStringsToInterfaces(eventIds))
 	if edgeXerr != nil {
 		return nil, nil, edgeXerr
 	}
 	e := models.Event{}
-	for _, pushedEvent := range pushedEvents {
-		err = json.Unmarshal(pushedEvent, &e)
+	for _, event := range events {
+		err = json.Unmarshal(event, &e)
 		if err != nil {
 			return nil, nil, errors.NewCommonEdgeX(errors.KindContractInvalid, "unable to marshal event", err)
 		}
 		rIds, err := redis.Strings(conn.Do(ZRANGE, fmt.Sprintf("%s:%s", EventsCollectionReadings, e.Id), 0, -1))
 		if err != nil {
-			return nil, nil, errors.NewCommonEdgeX(errors.KindDatabaseError, fmt.Sprintf("retrieve all reading Ids of pushed event %s failed", e.Id), err)
+			return nil, nil, errors.NewCommonEdgeX(errors.KindDatabaseError, fmt.Sprintf("retrieve all reading Ids of event %s failed", e.Id), err)
 		}
 		readingIds = append(readingIds, rIds...)
 	}
-	return pushedEventIds, readingIds, nil
+	return eventIds, readingIds, nil
 }
 
 func eventById(conn redis.Conn, id string) (event models.Event, edgeXerr errors.EdgeX) {