Repository: edgexfoundry/edgex-go (go)

Issue:
Allow default retention policy can apply to events not coming from auto events
<!--🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅

Hello there! 😄

To expedite issue processing please search open and closed issues before submitting a new one.
Existing issues often contain information about workarounds, resolution, or progress updates.

Please do not remove sections of this template when submitting your issue.  If the section is not relevant to your issue, leave the section but enter "n/a" for not applicable.

Note some sections are marked "REQUIRED" and must be provided.  Never mark required sections "n/a".

🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅🔅-->


# 🐞 Bug Report

### Affected Services [**REQUIRED**]
<!-- Can you pin-point one or more EdgeX services (Data, Metadata, Command, etc...) as the source of the bug? -->
<!-- ✍️edit: --> The issue is located in: Core Data


### Is this a regression?

<!-- Did this behavior work correctly in the previous version? -->
<!-- ✍️--> NO


### Description and Minimal Reproduction [**REQUIRED**]
The current default retention policy will only apply to events coming from auto events, which causes those events not coming from auto events remain in the database. Some device services, such as device-rest and device-websocket, don’t have the concept of auto events, so we should allow default retention policy can apply to these events.

## 🔥 Exception or Error
<pre><code>
<!-- If the issue is accompanied by an exception or an error, please share it below: -->
<!-- ✍️-->

</code></pre>


## 🌍  Your Environment
**Deployment Environment:**

**EdgeX Version [**REQUIRED**]:**


**Anything else relevant?**



Pull request description:
fix: CoreData apply default retention policy to none autoEvent
CoreData apply default retention policy to events not coming from auto events

Close #5188

<!-- Expected Commit Message Description (imported automatically by GitHub) -->
<!-- Must conform to [conventional commits guidelines](https://github.com/edgexfoundry/edgex-go/blob/main/.github/Contributing.md) -->
<!-- Expected Commit message must contain Closes/Fixes #IssueNumber statement when there is a related issue -->

<!-- Add additional detailed description of need for change if no related issue -->

**If your build fails**  due to your commit message not passing the build checks, please review the guidelines here: https://github.com/edgexfoundry/edgex-go/blob/main/.github/Contributing.md

## PR Checklist
Please check if your PR fulfills the following requirements:

- [x] I am not introducing a breaking change (if you are, flag in conventional commit message with `BREAKING CHANGE:` describing the break)
- [x] I am not introducing a new dependency (add notes below if you are)
- [x] I have added unit tests for the new feature or bug fix (if not, why?)
- [x] I have fully tested (add details below) this the new feature or bug fix (if not, why?)
- [ ] I have opened a PR for the related docs change (if not, why?)
  <link to docs PR>

## Testing Instructions
<!-- How can the reviewers test your change? -->
Deploy core service and device-virtual to test the event purge.

## New Dependency Instructions (If applicable)
<!-- Please follow [vetting instructions](https://wiki.edgexfoundry.org/display/FA/Vetting+Process+for+3rd+Party+Dependencies) and place results here -->

Patch:
diff --git a/internal/core/data/application/reading.go b/internal/core/data/application/reading.go
index ed219356..6e69600c 100644
--- a/internal/core/data/application/reading.go
+++ b/internal/core/data/application/reading.go
@@ -8,11 +8,13 @@ package application
 import (
 	"context"
 	"fmt"
+	"slices"
 	"sync"
 	"time"
 
 	bootstrapContainer "github.com/edgexfoundry/go-mod-bootstrap/v4/bootstrap/container"
 	"github.com/edgexfoundry/go-mod-bootstrap/v4/di"
+	"github.com/edgexfoundry/go-mod-core-contracts/v4/common"
 	"github.com/edgexfoundry/go-mod-core-contracts/v4/dtos"
 	"github.com/edgexfoundry/go-mod-core-contracts/v4/errors"
 	"github.com/edgexfoundry/go-mod-core-contracts/v4/models"
@@ -311,7 +313,7 @@ func AsyncPurgeEvent(ctx context.Context, dic *di.Container) errors.EdgeX {
 					lc.Info("Exiting event retention")
 					return
 				case <-timer.C:
-					err = purgeEventByAutoEvents(dic)
+					err = purgeEvents(dic)
 					if err != nil {
 						lc.Errorf("Failed to purge events and readings, %v", err)
 						break
@@ -324,20 +326,24 @@ func AsyncPurgeEvent(ctx context.Context, dic *di.Container) errors.EdgeX {
 	return nil
 }
 
-func purgeEventByAutoEvents(dic *di.Container) errors.EdgeX {
+func purgeEvents(dic *di.Container) errors.EdgeX {
 	devices := container.DeviceStoreFrom(dic.Get).Devices()
 	for _, device := range devices {
 		for _, e := range device.AutoEvents {
-			err := purgeEvent(device.Name, e, dic)
-			if err != nil {
+			if err := purgeEventsByAutoEvent(device.Name, e, dic); err != nil {
 				return errors.NewCommonEdgeXWrapper(err)
 			}
 		}
+
+		// purge events that are not in auto events
+		if err := purgeNoneAutoEventsByDevice(device, dic); err != nil {
+			return errors.NewCommonEdgeX(errors.Kind(err), fmt.Sprintf("purge device '%s' events not coming from auto events", device.Name), err)
+		}
 	}
 	return nil
 }
 
-func purgeEvent(deviceName string, autoEvent models.AutoEvent, dic *di.Container) errors.EdgeX {
+func purgeEventsByAutoEvent(deviceName string, autoEvent models.AutoEvent, dic *di.Container) errors.EdgeX {
 	config := container.ConfigurationFrom(dic.Get)
 	// apply the default retention policy, when maxCap/minCap/duration are not specified or equal to zero/empty string
 	// which are mentioned in the documentation
@@ -465,3 +471,58 @@ func countBasedEventRetention(deviceName string, autoEvent models.AutoEvent, dic
 	}
 	return nil
 }
+
+func purgeNoneAutoEventsByDevice(device models.Device, dic *di.Container) errors.EdgeX {
+	if len(device.ProfileName) == 0 {
+		return nil
+	}
+	sources, err := noneAutoEventSourcesByDevice(device, dic)
+	if err != nil {
+		return errors.NewCommonEdgeXWrapper(err)
+	}
+	for _, source := range sources {
+		autoEvent := models.AutoEvent{SourceName: source} // Empty AutoEvent.Retention will apply default retention policy
+		err = purgeEventsByAutoEvent(device.Name, autoEvent, dic)
+		if err != nil {
+			return errors.NewCommonEdgeXWrapper(err)
+		}
+	}
+	return nil
+}
+
+// noneAutoEventSourcesByDevice returns sources that are not defined in the device's auto events
+func noneAutoEventSourcesByDevice(device models.Device, dic *di.Container) ([]string, errors.EdgeX) {
+	profileClient := bootstrapContainer.DeviceProfileClientFrom(dic.Get)
+	profile, err := profileClient.DeviceProfileByName(context.Background(), device.ProfileName)
+	if err != nil {
+		return nil, errors.NewCommonEdgeX(errors.Kind(err), fmt.Sprintf("retrieve event sources not in auto events from the device '%s'", device.Name), err)
+	}
+	var sources []string
+	for _, r := range profile.Profile.DeviceResources {
+		// skip the device resource that is associated with any AutoEvent definitions.
+		if contained := slices.ContainsFunc(device.AutoEvents, func(event models.AutoEvent) bool {
+			return event.SourceName == r.Name
+		}); contained {
+			continue
+		}
+		// skip write-only device resource that do not generate events
+		if r.Properties.ReadWrite == common.ReadWrite_W {
+			continue
+		}
+		sources = append(sources, r.Name)
+	}
+	for _, c := range profile.Profile.DeviceCommands {
+		// skip the device command that is associated with any AutoEvent definitions.
+		if contained := slices.ContainsFunc(device.AutoEvents, func(event models.AutoEvent) bool {
+			return event.SourceName == c.Name
+		}); contained {
+			continue
+		}
+		// skip write-only device command that do not generate events
+		if c.ReadWrite == common.ReadWrite_W {
+			continue
+		}
+		sources = append(sources, c.Name)
+	}
+	return sources, nil
+}
diff --git a/internal/pkg/cache/device.go b/internal/pkg/cache/device.go
index e879f404..08b18c2d 100644
--- a/internal/pkg/cache/device.go
+++ b/internal/pkg/cache/device.go
@@ -3,7 +3,6 @@
 package cache
 
 import (
-	"maps"
 	"sync"
 
 	bootstrapContainer "github.com/edgexfoundry/go-mod-bootstrap/v4/bootstrap/container"
@@ -75,7 +74,11 @@ func (s *activeDeviceStore) Devices() map[string]models.Device {
 	s.mutex.RLock()
 	defer s.mutex.RUnlock()
 	// since the device map might change, return the copy of devices to prevent unexpected result
-	return maps.Clone(s.activeDeviceMap)
+	clonedDevices := make(map[string]models.Device, len(s.activeDeviceMap))
+	for deviceName, device := range s.activeDeviceMap {
+		clonedDevices[deviceName] = device.Clone()
+	}
+	return clonedDevices
 }
 
 func DeviceStore(dic *di.Container) ActiveDeviceStore {


Test patch:
diff --git a/internal/core/data/application/reading_test.go b/internal/core/data/application/reading_test.go
index 8c2cd03c..93083255 100644
--- a/internal/core/data/application/reading_test.go
+++ b/internal/core/data/application/reading_test.go
@@ -7,6 +7,7 @@
 package application
 
 import (
+	"context"
 	"net/http"
 	"testing"
 	"time"
@@ -19,7 +20,12 @@ import (
 	"github.com/edgexfoundry/edgex-go/internal/core/data/container"
 	dbMock "github.com/edgexfoundry/edgex-go/internal/core/data/infrastructure/interfaces/mocks"
 	"github.com/edgexfoundry/edgex-go/internal/core/data/mocks"
+	bootstrapContainer "github.com/edgexfoundry/go-mod-bootstrap/v4/bootstrap/container"
 	"github.com/edgexfoundry/go-mod-bootstrap/v4/di"
+	clientMocks "github.com/edgexfoundry/go-mod-core-contracts/v4/clients/interfaces/mocks"
+	"github.com/edgexfoundry/go-mod-core-contracts/v4/common"
+	"github.com/edgexfoundry/go-mod-core-contracts/v4/dtos"
+	"github.com/edgexfoundry/go-mod-core-contracts/v4/dtos/responses"
 	"github.com/edgexfoundry/go-mod-core-contracts/v4/errors"
 	"github.com/edgexfoundry/go-mod-core-contracts/v4/models"
 )
@@ -253,7 +259,7 @@ func TestReadingCountByDeviceName(t *testing.T) {
 	}
 }
 
-func TestPurgeEvent(t *testing.T) {
+func TestPurgeEventsByAutoEvent(t *testing.T) {
 	dic := mocks.NewMockDIC()
 	deviceName := "testDevice"
 	sourceName := "testSource"
@@ -340,7 +346,7 @@ func TestPurgeEvent(t *testing.T) {
 	}
 	for _, testCase := range tests {
 		t.Run(testCase.name, func(t *testing.T) {
-			err := purgeEvent(deviceName, testCase.autoEvent, dic)
+			err := purgeEventsByAutoEvent(deviceName, testCase.autoEvent, dic)
 			require.NoError(t, err)
 			duration, parseErr := time.ParseDuration(testCase.autoEvent.Retention.Duration)
 			require.NoError(t, parseErr)
@@ -363,3 +369,36 @@ func TestPurgeEvent(t *testing.T) {
 		})
 	}
 }
+
+func TestNoneAutoEventSourcesByDevice(t *testing.T) {
+	deviceName := "test-device"
+	profileName := "test-profile"
+	dic := mocks.NewMockDIC()
+	clientMock := &clientMocks.DeviceProfileClient{}
+	clientMock.On("DeviceProfileByName", context.Background(), profileName).Return(responses.DeviceProfileResponse{Profile: dtos.DeviceProfile{
+		DeviceResources: []dtos.DeviceResource{{Name: "resource1", Properties: dtos.ResourceProperties{ReadWrite: common.ReadWrite_R}}, {Name: "resource2", Properties: dtos.ResourceProperties{ReadWrite: common.ReadWrite_W}}},
+		DeviceCommands:  []dtos.DeviceCommand{{Name: "command1", ReadWrite: common.ReadWrite_R}, {Name: "command2", ReadWrite: common.ReadWrite_W}},
+	}}, nil)
+	dic.Update(di.ServiceConstructorMap{
+		bootstrapContainer.DeviceProfileClientName: func(get di.Get) interface{} {
+			return clientMock
+		},
+	})
+
+	test := []struct {
+		name            string
+		device          models.Device
+		expectedSources []string
+	}{
+		{"no auto events in device", models.Device{Name: deviceName, ProfileName: profileName, AutoEvents: nil}, []string{"resource1", "command1"}},
+		{"one auto events in device", models.Device{Name: deviceName, ProfileName: profileName, AutoEvents: []models.AutoEvent{{SourceName: "resource1"}}}, []string{"command1"}},
+		{"two auto events in device", models.Device{Name: deviceName, ProfileName: profileName, AutoEvents: []models.AutoEvent{{SourceName: "resource1"}, {SourceName: "command1"}}}, nil},
+	}
+	for _, testCase := range test {
+		t.Run(testCase.name, func(t *testing.T) {
+			sources, err := noneAutoEventSourcesByDevice(testCase.device, dic)
+			require.NoError(t, err)
+			assert.Equal(t, testCase.expectedSources, sources)
+		})
+	}
+}