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

Implement DELETE /event/device/{deviceName} 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.

Fix edgexfoundry#2837

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

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

@@ -182,3 +182,17 @@ func UpdateEventPushedById(id string, dic *di.Container) errors.EdgeX {
 		return nil
 	}
 }
+
+// The DeleteEventsByDeviceName function will be invoked by controller functions
+// and then invokes DeletePushedEvents function in the infrastructure layer to remove
+// all events that are associated with the given deviceName
+func DeleteEventsByDeviceName(deviceName string, dic *di.Container) errors.EdgeX {
+	dbClient := v2DataContainer.DBClientFrom(dic.Get)
+
+	err := dbClient.DeleteEventsByDeviceName(deviceName)
+	if err != nil {
+		return errors.NewCommonEdgeXWrapper(err)
+	}
+
+	return nil
+}

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,17 @@ 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
+		},
+	})
+
+	err := DeleteEventsByDeviceName(testDeviceName, dic)
+	require.NoError(t, err)
+}

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

@@ -223,7 +223,7 @@ func (ec *EventController) DeletePushedEvents(w http.ResponseWriter, r *http.Req
 	var response interface{}
 	var statusCode int
 
-	// Delete all events
+	// Delete all pushed events
 	err := application.DeletePushedEvents(ec.dic)
 	if err != nil {
 		lc.Debug(err.DebugMessages(), clients.CorrelationHeader, correlationId)
@@ -278,3 +278,33 @@ func (ec *EventController) UpdateEventPushedById(w http.ResponseWriter, r *http.
 	// encode and send out the response
 	pkg.Encode(updatedResponses, 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)
+
+	// URL parameters
+	vars := mux.Vars(r)
+	deviceName := vars[contractsV2.DeviceName]
+
+	var response interface{}
+	var statusCode int
+
+	// Delete events with associated Device name
+	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.ApiEventByDeviceRoute, http.NoBody)
+	req = mux.SetURLVars(req, map[string]string{v2.DeviceName: 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

@@ -20,4 +20,5 @@ type DBClient interface {
 	EventCountByDevice(deviceName string) (uint32, errors.EdgeX)
 	DeletePushedEvents() errors.EdgeX
 	UpdateEventPushedById(id string) errors.EdgeX
+	DeleteEventsByDeviceName(deviceName string) errors.EdgeX
 }

internal/core/data/v2/router.go

@@ -32,6 +32,7 @@ func LoadRestRoutes(r *mux.Router, dic *di.Container) {
 	r.HandleFunc(v2Constant.ApiEventCountByDeviceRoute, ec.EventCountByDevice).Methods(http.MethodGet)
 	r.HandleFunc(v2Constant.ApiEventPushRoute, ec.UpdateEventPushedById).Methods(http.MethodPut)
 	r.HandleFunc(v2Constant.ApiEventScrubRoute, ec.DeletePushedEvents).Methods(http.MethodDelete)
+	r.HandleFunc(v2Constant.ApiEventByDeviceRoute, ec.DeleteEventsByDeviceName).Methods(http.MethodDelete)
 
 	r.Use(correlation.ManageHeader)
 	r.Use(correlation.OnResponseComplete)

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

@@ -301,7 +301,7 @@ func (c *Client) DeletePushedEvents() errors.EdgeX {
 	conn := c.Pool.Get()
 	defer conn.Close()
 
-	pushedEventIds, readingIds, err := getPushedEventReadingIds(conn)
+	pushedEventIds, readingIds, err := getEventReadingIdsByKey(conn, EventsCollectionPushed)
 	if err != nil {
 		return errors.NewCommonEdgeXWrapper(err)
 	}
@@ -404,3 +404,17 @@ func (c *Client) DeviceByName(name string) (device model.Device, edgeXerr errors
 
 	return
 }
+
+// Delete all readings and events that are associated with the given device
+func (c *Client) DeleteEventsByDeviceName(deviceName string) 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)
+	}
+	deleteReadingsByIdChannel <- readingIds
+	deleteEventsByIdChannel <- eventIds
+	return nil
+}

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

@@ -321,3 +321,27 @@ func getPushedEventReadingIds(conn redis.Conn) (eventIds []string, readingIds []
 	}
 	return pushedEventIds, readingIds, nil
 }
+
+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, fmt.Sprintf("retrieve event ids by key %s failed", key), err)
+	}
+	events, edgeXerr := getObjectsByIds(conn, common.ConvertStringsToInterfaces(eventIds))
+	if edgeXerr != nil {
+		return nil, nil, edgeXerr
+	}
+	e := models.Event{}
+	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 event %s failed", e.Id), err)
+		}
+		readingIds = append(readingIds, rIds...)
+	}
+	return eventIds, readingIds, nil
+}

openapi/v2/core-data.yaml

@@ -910,15 +910,14 @@ paths:
               examples:
                 500Example:
                   $ref: '#/components/examples/500Example'
-  /event/device/{deviceId}:
+  /event/device/{deviceName}:
     parameters:
     - $ref: '#/components/parameters/correlatedRequestHeader'
-    - name: deviceId
+    - name: deviceName
       in: path
       required: true
       schema:
         type: string
-        format: uuid
       description: "Uniquely identifies a given device"
     delete:
       summary: "Deletes all events for the specified device"