Delete old redfish subscriptions (#49)

When hardware is moved the bmc may retain its old subscriptions.
These old subscriptions refer to the old location/xname. This causes
problems when the hardware sends events. The listeners mistakenly
think the events are for the old location/xname.

This commit changes hmcollector to automatically delete these
old subscriptions.

Here are the cases where it can find a bad subscription
- At start up it will check all subscriptions.
- Periodically it will check for its own subscriptions, and it will
delete any bad ones that it finds, however, in this case it does not
necessarily read all the subscriptions. It only reads until it finds
the subscription it is looking for.

CASMHMS-5685

.github/workflows/run_unit_test.yaml

@@ -0,0 +1,8 @@
+name: Run Unit Tests
+on: [push, pull_request, workflow_dispatch]
+jobs:
+  run_unit_test:
+    uses: Cray-HPE/hms-build-image-workflows/.github/workflows/run_unit_test.yaml@v2
+    with:
+      runs-on: ubuntu-latest
+    secrets: inherit

.version

@@ -1 +1 @@
-2.25.0
+2.26.0

CHANGELOG.md

@@ -23,6 +23,11 @@ Removed - for now removed features
 Fixed - for any bug fixes
 Security - in case of vulnerabilities
 -->
+## [2.26.0] - 2023-04-19
+
+### Changed
+- CASMHMS-5685 - Changed to prune event subscriptions with stale destinations. Stale subscriptions happen when hardware is physically moved.
+
 ## [2.25.0] - 2023-01-19
 
 ### Changed

Dockerfile

@@ -82,6 +82,7 @@ ENV KAFKA_HOST=localhost
 ENV KAFKA_PORT=9092
 ENV POLLING_INTERVAL=1
 ENV PDU_POLLING_INTERVAL=30
+ENV PRUNE_OLD_SUBSCRIPTIONS=true
 ENV HSM_REFRESH_INTERVAL=30
 ENV SM_URL=https://cray-smd
 ENV REST_URL=localhost

Dockerfile.testing

@@ -0,0 +1,55 @@
+# MIT License
+#
+# (C) Copyright [2023] Hewlett Packard Enterprise Development LP
+#
+# Permission is hereby granted, free of charge, to any person obtaining a
+# copy of this software and associated documentation files (the "Software"),
+# to deal in the Software without restriction, including without limitation
+# the rights to use, copy, modify, merge, publish, distribute, sublicense,
+# and/or sell copies of the Software, and to permit persons to whom the
+# Software is furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included
+# in all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+# THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR
+# OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
+# ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
+# OTHER DEALINGS IN THE SOFTWARE.
+
+# Dockerfile for running HMS Redfish Collector unit tests.
+
+# v1.1.0 of Confluent Go Kafka requires at least v1.1.0 of librdkafka.
+ARG LIBRDKAFKA_VER_MIN=1.1.0
+
+# Build base just has the packages installed we need.
+FROM artifactory.algol60.net/docker.io/library/golang:1.16-alpine AS build-base
+
+ARG LIBRDKAFKA_VER_MIN
+
+RUN set -ex \
+    && apk -U upgrade \
+    && apk add --no-cache \
+        build-base \
+        "librdkafka-dev>${LIBRDKAFKA_VER_MIN}" \
+        pkgconf
+
+# Base copies in the files we need to test/build.
+FROM build-base AS base
+
+RUN go env -w GO111MODULE=auto
+
+# Copy all the necessary files to the image.
+COPY cmd        $GOPATH/src/github.com/Cray-HPE/hms-hmcollector/cmd
+COPY internal   $GOPATH/src/github.com/Cray-HPE/hms-hmcollector/internal
+COPY vendor     $GOPATH/src/github.com/Cray-HPE/hms-hmcollector/vendor
+
+### Build Stage ###
+FROM base AS builder
+
+RUN set -ex \
+    && go test -cover github.com/Cray-HPE/hms-hmcollector/cmd/hmcollector
+

Makefile

@@ -25,10 +25,13 @@ NAME ?= hms-hmcollector
 VERSION ?= $(shell cat .version)
 DOCKER_IMAGE ?= ${NAME}:${VERSION}
 
-all: image snyk
+all: image unittest snyk
 
 image:
 	docker build ${NO_CACHE} --pull ${DOCKER_ARGS} --tag '${NAME}:${VERSION}' .
 
+unittest:
+	docker build -f Dockerfile.testing ${NO_CACHE} --pull ${DOCKER_ARGS} --tag '${NAME}-unittest:${VERSION}' .
+
 snyk:
 	./runSnyk.sh

cmd/hmcollector/hmcollector.go

@@ -69,6 +69,8 @@ var (
 	kafkaBrokersConfigFile = flag.String("kafka_brokers_config", "configs/kafka_brokers.json",
 		"Path to the configuration file containing all of the Kafka brokers this collector should produce to.")
 
+	pruneOldSubscriptions = flag.Bool("prune_old_subscriptions", true, "Should it prune old subscriptions that contain the wrong xname when compared to the endpoint.")
+
 	pollingInterval    = flag.Int("polling_interval", 10, "The polling interval to use in seconds.")
 	pduPollingInterval = flag.Int("pdu_polling_interval", 30, "The polling interval to use for redfish PDUs in seconds.")
 	hsmRefreshInterval = flag.Int("hsm_refresh_interval", 30,
@@ -180,7 +182,7 @@ func doUpdateHSMEndpoints() {
 	logger.Info("HSM endpoint monitoring routine shutdown.")
 }
 
-func setupLogging() {
+func SetupLogging() {
 	logLevel := os.Getenv("LOG_LEVEL")
 	logLevel = strings.ToUpper(logLevel)
 
@@ -277,7 +279,7 @@ func caChangeCB(caBundle string) {
 func main() {
 	var err error
 
-	setupLogging()
+	SetupLogging()
 
 	serviceName, err = base.GetServiceInstanceName()
 	if err != nil {

cmd/hmcollector/rf_subscriptions.go

@@ -1,6 +1,6 @@
 // MIT License
 //
-// (C) Copyright [2020-2021] Hewlett Packard Enterprise Development LP
+// (C) Copyright [2020-2021,2023] Hewlett Packard Enterprise Development LP
 //
 // Permission is hereby granted, free of charge, to any person obtaining a
 // copy of this software and associated documentation files (the "Software"),
@@ -36,6 +36,7 @@ import (
 	base "github.com/Cray-HPE/hms-base"
 	"github.com/Cray-HPE/hms-hmcollector/internal/hmcollector"
 	rf "github.com/Cray-HPE/hms-smd/pkg/redfish"
+	"github.com/Cray-HPE/hms-xname/xnametypes"
 	"go.uber.org/zap"
 )
 
@@ -116,6 +117,104 @@ func postRFSubscription(endpoint *rf.RedfishEPDescription, evTypes []string, reg
 	return subscribed, err
 }
 
+func isSubscriptionForWrongXname(xname string, eventSub *hmcollector.EventSubscription) bool {
+	if !*pruneOldSubscriptions {
+		return false
+	}
+
+	if xnametypes.GetHMSType(xname) == xnametypes.HMSTypeInvalid {
+		logger.Debug("The endpoint ID is unexpectedly not an xname when checking for stale subscriptions.",
+			zap.Any("xname", xname))
+		// This function can only check for bad subscriptions when the xname passed to it is valid
+		return false
+	}
+
+	xnameOfSub := eventSub.Context
+	if xnametypes.GetHMSType(xnameOfSub) == xnametypes.HMSTypeInvalid {
+		destinationParts := strings.Split(eventSub.Destination, "/")
+		xnameOfSub = destinationParts[len(destinationParts)-1]
+		if xnametypes.GetHMSType(xnameOfSub) == xnametypes.HMSTypeInvalid {
+			// Neither the Destination nor the Context contain a valid xname
+			return false
+		}
+	}
+
+	xnameOfSubNormalized := xnametypes.NormalizeHMSCompID(xnameOfSub)
+	xnameOfEndpointNormalized := xnametypes.NormalizeHMSCompID(xname)
+	if xnameOfSubNormalized != xnameOfEndpointNormalized {
+		// The destination ends in a valid xname, or the context is a valid xname.
+		// The xname does not match the hostname/xname of the redfish endpoint.
+		// Conclusion:
+		//   The subscription was created by hms-hmcollector and not slingshot or something else.
+		//   The subscription is for the wrong endpoint
+		//   The hardware was likely physically moved and this old subscription should be deleted.
+		logger.Info("Found mismatched subscription",
+			zap.String("endpoint", xname),
+			zap.String("destination", eventSub.Destination),
+			zap.String("context", eventSub.Context),
+			zap.String("normalized endpoint xname", xnameOfEndpointNormalized),
+			zap.String("normalized subscription xname", xnameOfSubNormalized),
+		)
+		return true
+	}
+	return false
+}
+
+func deleteSubscriptionForWrongXname(endpoint *rf.RedfishEPDescription, subUri string, eventSub *hmcollector.EventSubscription) {
+	// This function should only be used to delete subscriptions where isSubscriptionForWrongXname returns true.
+	baseEndpointURL := "https://" + endpoint.FQDN
+	fullURL := baseEndpointURL + subUri
+
+	endpointLogger := logger.With(
+		zap.Any("xname", endpoint.ID),
+		zap.Any("destination", eventSub.Destination),
+		zap.Any("context", eventSub.Context),
+	)
+
+	endpointLogger.Warn("Subscription does not match endpoint. Deleting the subscription.")
+
+	deleteRetBytes, _, deleteErr := doHTTPAction(endpoint, http.MethodDelete, fullURL, nil)
+	if deleteErr != nil {
+		endpointLogger.Error("Failed to delete subscription!", zap.Error(deleteErr))
+	} else {
+		endpointLogger.Info("Deleted subscription.",
+			zap.String("string(deleteRetBytes)", string(deleteRetBytes)))
+	}
+}
+
+func getSubscriptions(endpoint *rf.RedfishEPDescription) (*hmcollector.EventSubscriptionCollection, error) {
+	baseEndpointURL := "https://" + endpoint.FQDN
+	fullURL := fmt.Sprintf("%s/redfish/v1/EventService/Subscriptions", baseEndpointURL)
+	payloadBytes, _, err := doHTTPAction(endpoint, http.MethodGet, fullURL, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	var subList hmcollector.EventSubscriptionCollection
+	err = json.Unmarshal(payloadBytes, &subList)
+	if err != nil {
+		return nil, err
+	}
+	return &subList, nil
+}
+
+func getSubscription(endpoint *rf.RedfishEPDescription, subUri string) (*hmcollector.EventSubscription, error) {
+	baseEndpointURL := "https://" + endpoint.FQDN
+	fullURL := baseEndpointURL + subUri
+
+	payloadBytes, _, err := doHTTPAction(endpoint, http.MethodGet, fullURL, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	var eventSub hmcollector.EventSubscription
+	err = json.Unmarshal(payloadBytes, &eventSub)
+	if err != nil {
+		return nil, err
+	}
+	return &eventSub, nil
+}
+
 func isDupRFSubscription(endpoint *rf.RedfishEPDescription, registryPrefixes []string) (bool, error) {
 	baseEndpointURL := "https://" + endpoint.FQDN
 
@@ -191,6 +290,11 @@ func isDupRFSubscription(endpoint *rf.RedfishEPDescription, registryPrefixes []s
 				// it attempted to delete the subscription, so return no match.
 				return match, nil
 			}
+		} else {
+			if *pruneOldSubscriptions &&
+				isSubscriptionForWrongXname(endpoint.ID, &eventSub) {
+				deleteSubscriptionForWrongXname(endpoint, sub.OId, &eventSub)
+			}
 		}
 	}
 	return false, nil
@@ -347,6 +451,27 @@ func rfSubscribe(pendingRFSubscriptions <-chan hmcollector.RFSub) {
 				registryPrefixGroups = append(registryPrefixGroups, []string{"CrayTelemetry"})
 			}
 
+			if *pruneOldSubscriptions {
+				subscriptions, err := getSubscriptions(sub.Endpoint)
+				if err != nil {
+					logger.Error("Unable to get subscriptions!", zap.String("xname", sub.Endpoint.ID), zap.Error(err))
+				} else {
+					for _, member := range subscriptions.Members {
+						eventSub, err := getSubscription(sub.Endpoint, member.OId)
+						if err != nil {
+							logger.Error("Unable to get subscription!",
+								zap.String("xname", sub.Endpoint.ID),
+								zap.String("id", member.OId),
+								zap.Error(err))
+							continue
+						}
+						if isSubscriptionForWrongXname(sub.Endpoint.ID, eventSub) {
+							deleteSubscriptionForWrongXname(sub.Endpoint, member.OId, eventSub)
+						}
+					}
+				}
+			}
+
 			// Set up a subscription for the required registry prefix groups.
 			for _, registryPrefixGroup := range registryPrefixGroups {
 				// Check the endpoint to see if we are already subscribed.

cmd/hmcollector/rf_subscriptions_test.go

@@ -0,0 +1,100 @@
+// MIT License
+//
+// (C) Copyright [2023] Hewlett Packard Enterprise Development LP
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the "Software"),
+// to deal in the Software without restriction, including without limitation
+// the rights to use, copy, modify, merge, publish, distribute, sublicense,
+// and/or sell copies of the Software, and to permit persons to whom the
+// Software is furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included
+// in all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
+// THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR
+// OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
+// ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
+// OTHER DEALINGS IN THE SOFTWARE.
+
+package main
+
+import (
+	"os"
+	"testing"
+
+	"github.com/Cray-HPE/hms-hmcollector/internal/hmcollector"
+)
+
+func TestMain(m *testing.M) {
+	setupLoggingForTest()
+	m.Run()
+}
+
+func setupLoggingForTest() {
+	logLevel := os.Getenv("LOG_LEVEL")
+	if logLevel == "" {
+		os.Setenv("LOG_LEVEL", "debug")
+	}
+	SetupLogging()
+}
+
+func TestTrueCaseForIsSubscriptionForWrongXname(t *testing.T) {
+	xname := "x3000c0s19b1"
+	wrongXnameSubscriptions := []*hmcollector.EventSubscription{
+		&hmcollector.EventSubscription{Destination: "http://10.94.100.71/x3000c0s19b7"},
+		&hmcollector.EventSubscription{Destination: "http://10.94.100.71", Context: "x3000c0s19b7"},
+	}
+	for i, sub := range wrongXnameSubscriptions {
+		if !isSubscriptionForWrongXname(xname, sub) {
+			t.Errorf("Expected isSubscriptionForWrongXname to return true. "+
+				"i: %d, xname: %s, destination: %s, context: %s",
+				i, xname, sub.Destination, sub.Context)
+		}
+	}
+}
+
+func TestFalseCaseForIsSubscriptionForWrongXname(t *testing.T) {
+	xname := "x3000c0s19b1"
+	unnormalizedXname := "x3000c0s19b000001"
+	subscriptions := []*hmcollector.EventSubscription{
+		&hmcollector.EventSubscription{Destination: "http://10.94.100.71/" + xname, Context: ""},
+		&hmcollector.EventSubscription{Destination: "http://10.94.100.71/", Context: xname},
+		&hmcollector.EventSubscription{Destination: "http://10.94.100.71", Context: xname},
+		&hmcollector.EventSubscription{Destination: "http://10.94.100.71/" + unnormalizedXname, Context: ""},
+		&hmcollector.EventSubscription{Destination: "http://10.94.100.71", Context: unnormalizedXname},
+		&hmcollector.EventSubscription{Destination: "http://10.94.100.71/" + unnormalizedXname, Context: unnormalizedXname},
+		// subscriptions that are not owned by hmcollector
+		&hmcollector.EventSubscription{Destination: "http://10.94.100.71", Context: "fabric-manager-hardware-telemetry-sub"},
+		&hmcollector.EventSubscription{Destination: "http://10.94.100.71:80", Context: "fabric-manager-hardware-telemetry-sub"},
+		&hmcollector.EventSubscription{Destination: "http://10.94.100.71", Context: ""},
+		&hmcollector.EventSubscription{Destination: "http://10.94.100.71/junk", Context: ""},
+		&hmcollector.EventSubscription{Destination: "", Context: ""},
+	}
+	for i, sub := range subscriptions {
+		if isSubscriptionForWrongXname(xname, sub) {
+			t.Errorf("Expected isSubscriptionForWrongXname to return false. "+
+				"i: %d, xname: %s, destination: %s, context: %s",
+				i, xname, sub.Destination, sub.Context)
+		}
+	}
+}
+func TestBadXnameCaseForIsSubscriptionForWrongXname(t *testing.T) {
+	xname := "10.94.101.71"
+	unnormalizedXname := "x3000c0s19b000001"
+	subscriptions := []*hmcollector.EventSubscription{
+		&hmcollector.EventSubscription{Destination: "http://" + xname, Context: ""},
+		&hmcollector.EventSubscription{Destination: "http://" + xname, Context: xname},
+		&hmcollector.EventSubscription{Destination: "http://" + xname, Context: unnormalizedXname},
+	}
+	for i, sub := range subscriptions {
+		if isSubscriptionForWrongXname(xname, sub) {
+			t.Errorf("Expected isSubscriptionForWrongXname to return false. "+
+				"i: %d, xname: %s, destination: %s, context: %s",
+				i, xname, sub.Destination, sub.Context)
+		}
+	}
+}

go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/Cray-HPE/hms-compcredentials v1.11.2
 	github.com/Cray-HPE/hms-securestorage v1.12.2
 	github.com/Cray-HPE/hms-smd v1.30.9
+	github.com/Cray-HPE/hms-xname v1.1.0
 	github.com/Shopify/sarama v1.23.1
 	github.com/eapache/go-resiliency v1.2.0 // indirect
 	github.com/namsral/flag v1.7.4-pre