add nodeName as a new parameter to support MNO usecase

pkg/clients/clientset.go

@@ -98,8 +98,13 @@ func ClearClientSet() {
 	clientset = Clientset{}
 }
 
-func (clientsholder *Clientset) FindPodNameFromPrefix(namespace, prefix string) (string, error) {
-	podList, err := clientsholder.K8sClient.CoreV1().Pods(namespace).List(context.TODO(), metav1.ListOptions{})
+func (clientsholder *Clientset) FindPodNameFromPrefix(namespace, prefix, nodeName string) (string, error) {
+	listOpts := metav1.ListOptions{}
+	if len(nodeName) > 0 {
+		listOpts = metav1.ListOptions{FieldSelector: "spec.nodeName=" + nodeName}
+	}
+	podList, err := clientsholder.K8sClient.CoreV1().Pods(namespace).List(context.TODO(), listOpts)
+
 	if err != nil {
 		return "", fmt.Errorf("failed to getting pod list: %w", err)
 	}
@@ -115,10 +120,12 @@ func (clientsholder *Clientset) FindPodNameFromPrefix(namespace, prefix string)
 
 	switch len(podNames) {
 	case 0:
-		return "", fmt.Errorf("no pod with prefix %v found in namespace %v", prefix, namespace)
+		return "", fmt.Errorf("no pod with prefix %v found in namespace %v on node %v", prefix, namespace,
+			nodeName)
 	case 1:
 		return podNames[0], nil
 	default:
-		return "", fmt.Errorf("too many (%v) pods with prefix %v found in namespace %v", len(podNames), prefix, namespace)
+		return "", fmt.Errorf("too many (%v) pods with prefix %v found in namespace %v on node %v",
+			len(podNames), prefix, namespace, nodeName)
 	}
 }

pkg/clients/exec_command.go

@@ -32,17 +32,18 @@ type ExecContext interface {
 
 var NewSPDYExecutor = remotecommand.NewSPDYExecutor
 
-// ContainerExecContext encapsulates the context in which a command is run; the namespace, pod, and container.
+// ContainerExecContext encapsulates the context in which a command is run; the nodeName, the namespace, pod, and container.
 type ContainerExecContext struct {
 	clientset     *Clientset
 	namespace     string
 	podName       string
 	containerName string
 	podNamePrefix string
+	nodeName      string
 }
 
 func (c *ContainerExecContext) refresh() error {
-	newPodname, err := c.clientset.FindPodNameFromPrefix(c.namespace, c.podNamePrefix)
+	newPodname, err := c.clientset.FindPodNameFromPrefix(c.namespace, c.podNamePrefix, c.nodeName)
 	if err != nil {
 		return err
 	}
@@ -52,9 +53,9 @@ func (c *ContainerExecContext) refresh() error {
 
 func NewContainerContext(
 	clientset *Clientset,
-	namespace, podNamePrefix, containerName string,
+	namespace, podNamePrefix, containerName, nodeName string,
 ) (*ContainerExecContext, error) {
-	podName, err := clientset.FindPodNameFromPrefix(namespace, podNamePrefix)
+	podName, err := clientset.FindPodNameFromPrefix(namespace, podNamePrefix, nodeName)
 	if err != nil {
 		return &ContainerExecContext{}, err
 	}
@@ -64,6 +65,7 @@ func NewContainerContext(
 		containerName: containerName,
 		podNamePrefix: podNamePrefix,
 		clientset:     clientset,
+		nodeName:      nodeName,
 	}
 	return &ctx, nil
 }

pkg/clients/exec_command_test.go

@@ -24,13 +24,19 @@ var notATestPod = &v1.Pod{
 		Namespace:   "TestNamespace",
 		Annotations: map[string]string{},
 	},
+	Spec: v1.PodSpec{
+		NodeName: "TestNode",
+	},
 }
 var testPod = &v1.Pod{
 	ObjectMeta: metav1.ObjectMeta{
 		Name:        "TestPod-8292",
 		Namespace:   "TestNamespace",
 		Annotations: map[string]string{},
 	},
+	Spec: v1.PodSpec{
+		NodeName: "TestNode",
+	},
 }
 
 var _ = Describe("NewContainerContext", func() {
@@ -49,7 +55,7 @@ var _ = Describe("NewContainerContext", func() {
 			fakeK8sClient := fakeK8s.NewSimpleClientset(notATestPod)
 			clientset.K8sClient = fakeK8sClient
 
-			_, err := clients.NewContainerContext(clientset, "TestNamespace", "Test", "TestContainer")
+			_, err := clients.NewContainerContext(clientset, "TestNamespace", "Test", "TestContainer", "TestNode")
 			Expect(err).To(HaveOccurred())
 		})
 	})
@@ -58,7 +64,7 @@ var _ = Describe("NewContainerContext", func() {
 			fakeK8sClient := fakeK8s.NewSimpleClientset(notATestPod, testPod)
 			clientset.K8sClient = fakeK8sClient
 
-			ctx, err := clients.NewContainerContext(clientset, "TestNamespace", "Test", "TestContainer")
+			ctx, err := clients.NewContainerContext(clientset, "TestNamespace", "Test", "TestContainer", "TestNode")
 			Expect(err).NotTo(HaveOccurred())
 			Expect(ctx.GetNamespace()).To(Equal("TestNamespace"))
 			Expect(ctx.GetContainerName()).To(Equal("TestContainer"))
@@ -81,7 +87,7 @@ var _ = Describe("ExecCommandContainer", func() {
 				return []byte(expectedStdOut), []byte(expectedStdErr), nil
 			}
 			clients.NewSPDYExecutor = testutils.NewFakeNewSPDYExecutor(responder, nil)
-			ctx, _ := clients.NewContainerContext(clientset, "TestNamespace", "Test", "TestContainer")
+			ctx, _ := clients.NewContainerContext(clientset, "TestNamespace", "Test", "TestContainer", "TestNode")
 			cmd := []string{"my", "test", "command"}
 			stdout, stderr, err := ctx.ExecCommand(cmd)
 			Expect(err).NotTo(HaveOccurred())
@@ -100,7 +106,7 @@ var _ = Describe("ExecCommandContainer", func() {
 				return []byte(expectedStdOut), []byte(expectedStdErr), nil
 			}
 			clients.NewSPDYExecutor = testutils.NewFakeNewSPDYExecutor(responder, expectedErr)
-			ctx, _ := clients.NewContainerContext(clientset, "TestNamespace", "Test", "TestContainer")
+			ctx, _ := clients.NewContainerContext(clientset, "TestNamespace", "Test", "TestContainer", "TestNode")
 			cmd := []string{"my", "test", "command"}
 			stdout, stderr, err := ctx.ExecCommand(cmd)
 			Expect(err).To(HaveOccurred())
@@ -119,7 +125,7 @@ var _ = Describe("ExecCommandContainer", func() {
 				return []byte(expectedStdOut), []byte(expectedStdErr), expectedErr
 			}
 			clients.NewSPDYExecutor = testutils.NewFakeNewSPDYExecutor(responder, nil)
-			ctx, _ := clients.NewContainerContext(clientset, "TestNamespace", "Test", "TestContainer")
+			ctx, _ := clients.NewContainerContext(clientset, "TestNamespace", "Test", "TestContainer", "TestNode")
 			cmd := []string{"my", "test", "command"}
 			stdout, stderr, err := ctx.ExecCommand(cmd)
 			Expect(err).To(HaveOccurred())

pkg/cmd/collect.go

@@ -81,6 +81,7 @@ var collectCmd = &cobra.Command{
 		collectionRunner.Run(
 			kubeConfig,
 			outputFile,
+			nodeName,
 			requestedDuration,
 			pollInterval,
 			devInfoAnnouceInterval,
@@ -101,6 +102,7 @@ func init() { //nolint:funlen // Allow this to get a little long
 	AddOutputFlag(collectCmd)
 	AddFormatFlag(collectCmd)
 	AddInterfaceFlag(collectCmd)
+	AddNodeNameFlag(collectCmd)
 
 	collectCmd.Flags().StringVarP(
 		&requestedDurationStr,

pkg/cmd/common.go

@@ -13,16 +13,23 @@ var (
 	outputFile      string
 	useAnalyserJSON bool
 	ptpInterface    string
+	nodeName        string
 )
 
 func AddKubeconfigFlag(targetCmd *cobra.Command) {
-	targetCmd.Flags().StringVarP(&kubeConfig, "kubeconfig", "k", "", "Path to the kubeconfig file")
+	targetCmd.Flags().StringVarP(&kubeConfig,
+		"kubeconfig",
+		"k", "",
+		"Path to the kubeconfig file")
 	err := targetCmd.MarkFlagRequired("kubeconfig")
 	utils.IfErrorExitOrPanic(err)
 }
 
 func AddOutputFlag(targetCmd *cobra.Command) {
-	targetCmd.Flags().StringVarP(&outputFile, "output", "o", "", "Path to the output file")
+	targetCmd.Flags().StringVarP(&outputFile,
+		"output",
+		"o", "",
+		"Path to the output file")
 }
 
 func AddFormatFlag(targetCmd *cobra.Command) {
@@ -36,7 +43,17 @@ func AddFormatFlag(targetCmd *cobra.Command) {
 }
 
 func AddInterfaceFlag(targetCmd *cobra.Command) {
-	targetCmd.Flags().StringVarP(&ptpInterface, "interface", "i", "", "Name of the PTP interface")
+	targetCmd.Flags().StringVarP(&ptpInterface,
+		"interface",
+		"i", "",
+		"Name of the PTP interface")
 	err := targetCmd.MarkFlagRequired("interface")
 	utils.IfErrorExitOrPanic(err)
 }
+
+func AddNodeNameFlag(targetCmd *cobra.Command) {
+	targetCmd.Flags().StringVarP(&ptpInterface,
+		"nodeName",
+		"n", "",
+		"Name of the Node under test (valid only for MNO Use case)")
+}

pkg/cmd/verifyEnv.go

@@ -20,7 +20,7 @@ var verifyEnvCmd = &cobra.Command{
 	Short: "verify the environment is ready for collection",
 	Long:  `verify the environment is ready for collection`,
 	Run: func(cmd *cobra.Command, args []string) {
-		verify.Verify(ptpInterface, kubeConfig, useAnalyserJSON)
+		verify.Verify(ptpInterface, kubeConfig, useAnalyserJSON, nodeName)
 	},
 }
 
@@ -31,4 +31,5 @@ func init() {
 	AddOutputFlag(verifyEnvCmd)
 	AddFormatFlag(verifyEnvCmd)
 	AddInterfaceFlag(verifyEnvCmd)
+	AddNodeNameFlag(verifyEnvCmd)
 }

pkg/collectors/collector.go

@@ -24,6 +24,7 @@ type CollectionConstructor struct {
 	Clientset              *clients.Clientset
 	ErroredPolls           chan PollResult
 	PTPInterface           string
+	PTPNodeName            string
 	Msg                    string
 	LogsOutputFile         string
 	TempDir                string

pkg/collectors/contexts/contexts.go

@@ -20,8 +20,8 @@ const (
 	NetlinkDebugContainerImage = "quay.io/redhat-partner-solutions/dpll-debug:0.1"
 )
 
-func GetPTPDaemonContext(clientset *clients.Clientset) (clients.ExecContext, error) {
-	ctx, err := clients.NewContainerContext(clientset, PTPNamespace, PTPPodNamePrefix, PTPContainer)
+func GetPTPDaemonContext(clientset *clients.Clientset, ptpNodeName string) (clients.ExecContext, error) {
+	ctx, err := clients.NewContainerContext(clientset, PTPNamespace, PTPPodNamePrefix, PTPContainer, ptpNodeName)
 	if err != nil {
 		return ctx, fmt.Errorf("could not create container context %w", err)
 	}

pkg/collectors/dev_info_collector.go

@@ -151,7 +151,7 @@ func verify(ptpDevInfo *devices.PTPDeviceInfo, constructor *CollectionConstructo
 // Returns a new DevInfoCollector from the CollectionConstuctor Factory
 func NewDevInfoCollector(constructor *CollectionConstructor) (Collector, error) {
 	// Build DPPInfoFetcher ahead of time call to GetPTPDeviceInfo will build the other
-	ctx, err := contexts.GetPTPDaemonContext(constructor.Clientset)
+	ctx, err := contexts.GetPTPDaemonContext(constructor.Clientset, constructor.PTPNodeName)
 	if err != nil {
 		return &DevInfoCollector{}, fmt.Errorf("failed to create DevInfoCollector: %w", err)
 	}

pkg/collectors/devices/device_info_test.go

@@ -40,6 +40,9 @@ var testPod = &v1.Pod{
 		Namespace:   "TestNamespace",
 		Annotations: map[string]string{},
 	},
+	Spec: v1.PodSpec{
+		NodeName: "TestNodeName",
+	},
 }
 
 var _ = Describe("NewContainerContext", func() {
@@ -84,7 +87,7 @@ var _ = Describe("NewContainerContext", func() {
 
 			response[expectedInput] = []byte(expectedOutput)
 
-			ctx, err := clients.NewContainerContext(clientset, "TestNamespace", "Test", "TestContainer")
+			ctx, err := clients.NewContainerContext(clientset, "TestNamespace", "Test", "TestContainer", "TestNodeName")
 			Expect(err).NotTo(HaveOccurred())
 			info, err := devices.GetPTPDeviceInfo("aFakeInterface", ctx)
 			Expect(err).NotTo(HaveOccurred())

pkg/collectors/devices/dpll_fs_test.go

@@ -54,7 +54,7 @@ var _ = Describe("NewContainerContext", func() {
 
 			response[expectedInput] = []byte(expectedOutput)
 
-			ctx, err := clients.NewContainerContext(clientset, "TestNamespace", "Test", "TestContainer")
+			ctx, err := clients.NewContainerContext(clientset, "TestNamespace", "Test", "TestContainer", "TestNodeName")
 			Expect(err).NotTo(HaveOccurred())
 			info, err := devices.GetDevDPLLFilesystemInfo(ctx, "aFakeInterface")
 			Expect(err).NotTo(HaveOccurred())

pkg/collectors/devices/dpll_netlink.go

@@ -187,8 +187,8 @@ func postProcessDPLLNetlinkClockID(result map[string]string) (map[string]any, er
 }
 
 type NetlinkClockID struct {
-	ClockID   *big.Int `fetcherKey:"clockID"       json:"clockId"`
-	Timestamp string   `fetcherKey:"date"          json:"timestamp"`
+	ClockID   *big.Int `fetcherKey:"clockID" json:"clockId"`
+	Timestamp string   `fetcherKey:"date"    json:"timestamp"`
 }
 
 func GetClockID(ctx clients.ExecContext, interfaceName string) (NetlinkClockID, error) {

pkg/collectors/devices/gps_ubx_test.go

@@ -65,7 +65,7 @@ var _ = Describe("GetGPSNav", func() {
 			}, "\n")
 			response[expectedInput] = []byte(expectedOutput)
 
-			ctx, err := clients.NewContainerContext(clientset, "TestNamespace", "Test", "TestContainer")
+			ctx, err := clients.NewContainerContext(clientset, "TestNamespace", "Test", "TestContainer", "TestNodeName")
 			Expect(err).NotTo(HaveOccurred())
 
 			gpsInfo, err := devices.GetGPSNav(ctx)

pkg/collectors/devices/gps_ubx_ver_test.go

@@ -72,7 +72,7 @@ var _ = Describe("GetGPSNav", func() {
 			}, "\n")
 			response[expectedInput] = []byte(expectedOutput)
 
-			ctx, err := clients.NewContainerContext(clientset, "TestNamespace", "Test", "TestContainer")
+			ctx, err := clients.NewContainerContext(clientset, "TestNamespace", "Test", "TestContainer", "TestNodeName")
 			Expect(err).NotTo(HaveOccurred())
 
 			gpsInfo, err := devices.GetGPSVersions(ctx)

pkg/collectors/devices/pmc_test.go

@@ -64,7 +64,7 @@ var _ = Describe("GetPMC", func() {
 			}, "\n")
 			response[expectedInput] = []byte(expectedOutput)
 
-			ctx, err := clients.NewContainerContext(clientset, "TestNamespace", "Test", "TestContainer")
+			ctx, err := clients.NewContainerContext(clientset, "TestNamespace", "Test", "TestContainer", "TestNodeName")
 			Expect(err).NotTo(HaveOccurred())
 
 			pmcInfo, err := devices.GetPMC(ctx)

pkg/collectors/dpll_collector.go

@@ -17,7 +17,7 @@ const (
 
 // Returns a new DPLLCollector from the CollectionConstuctor Factory
 func NewDPLLCollector(constructor *CollectionConstructor) (Collector, error) {
-	ctx, err := contexts.GetPTPDaemonContext(constructor.Clientset)
+	ctx, err := contexts.GetPTPDaemonContext(constructor.Clientset, constructor.PTPNodeName)
 	if err != nil {
 		return &DPLLNetlinkCollector{}, fmt.Errorf("failed to create DPLLCollector: %w", err)
 	}

pkg/collectors/dpll_collector_fs.go

@@ -67,7 +67,7 @@ func (dpll *DPLLFilesystemCollector) CleanUp() error {
 
 // Returns a new DPLLFilesystemCollector from the CollectionConstuctor Factory
 func NewDPLLFilesystemCollector(constructor *CollectionConstructor) (Collector, error) {
-	ctx, err := contexts.GetPTPDaemonContext(constructor.Clientset)
+	ctx, err := contexts.GetPTPDaemonContext(constructor.Clientset, constructor.PTPNodeName)
 	if err != nil {
 		return &DPLLFilesystemCollector{}, fmt.Errorf("failed to create DPLLFilesystemCollector: %w", err)
 	}

pkg/collectors/gps_ubx_collector.go

@@ -54,7 +54,7 @@ func (gps *GPSCollector) Poll(resultsChan chan PollResult, wg *utils.WaitGroupCo
 
 // Returns a new GPSCollector based on values in the CollectionConstructor
 func NewGPSCollector(constructor *CollectionConstructor) (Collector, error) {
-	ctx, err := contexts.GetPTPDaemonContext(constructor.Clientset)
+	ctx, err := contexts.GetPTPDaemonContext(constructor.Clientset, constructor.PTPNodeName)
 	if err != nil {
 		return &GPSCollector{}, fmt.Errorf("failed to create DPLLCollector: %w", err)
 	}

pkg/collectors/log_follower.go

@@ -63,6 +63,7 @@ type LogsCollector struct {
 	client             *clients.Clientset
 	sliceQuit          chan os.Signal
 	logsOutputFileName string
+	nodeName           string
 	lastPoll           loglines.GenerationalLockedTime
 	wg                 sync.WaitGroup
 	withTimeStamps     bool
@@ -200,7 +201,7 @@ func processStream(stream io.ReadCloser, expectedEndtime time.Time) ([]*loglines
 }
 
 func (logs *LogsCollector) poll() error {
-	podName, err := logs.client.FindPodNameFromPrefix(contexts.PTPNamespace, contexts.PTPPodNamePrefix)
+	podName, err := logs.client.FindPodNameFromPrefix(contexts.PTPNamespace, contexts.PTPPodNamePrefix, logs.nodeName)
 	if err != nil {
 		return fmt.Errorf("failed to poll: %w", err)
 	}
@@ -282,6 +283,7 @@ func NewLogsCollector(constructor *CollectionConstructor) (Collector, error) {
 			Store:  make(map[uint32][]*loglines.LineSlice),
 			Dumper: loglines.NewGenerationDumper(constructor.TempDir, constructor.KeepDebugFiles),
 		},
+		nodeName: constructor.PTPNodeName,
 	}
 	return &collector, nil
 }

pkg/collectors/pmc_collector.go

@@ -53,7 +53,7 @@ func (pmc *PMCCollector) Poll(resultsChan chan PollResult, wg *utils.WaitGroupCo
 
 // Returns a new PMCCollector based on values in the CollectionConstructor
 func NewPMCCollector(constructor *CollectionConstructor) (Collector, error) {
-	ctx, err := contexts.GetPTPDaemonContext(constructor.Clientset)
+	ctx, err := contexts.GetPTPDaemonContext(constructor.Clientset, constructor.PTPNodeName)
 	if err != nil {
 		return &PMCCollector{}, fmt.Errorf("failed to create PMCCollector: %w", err)
 	}

pkg/runner/runner.go

@@ -63,6 +63,7 @@ func NewCollectorRunner(selectedCollectors []string) *CollectorRunner {
 func (runner *CollectorRunner) initialise( //nolint:funlen // allow a slightly long function
 	callback callbacks.Callback,
 	ptpInterface string,
+	ptpNodeName string,
 	clientset *clients.Clientset,
 	pollInterval int,
 	requestedDuration time.Duration,
@@ -79,6 +80,7 @@ func (runner *CollectorRunner) initialise( //nolint:funlen // allow a slightly l
 	constructor := &collectors.CollectionConstructor{
 		Callback:               callback,
 		PTPInterface:           ptpInterface,
+		PTPNodeName:            ptpNodeName,
 		Clientset:              clientset,
 		PollInterval:           pollInterval,
 		DevInfoAnnouceInterval: devInfoAnnouceInterval,
@@ -220,6 +222,7 @@ func (runner *CollectorRunner) cleanUpAll() {
 func (runner *CollectorRunner) Run( //nolint:funlen // allow a slightly long function
 	kubeConfig string,
 	outputFile string,
+	nodeName string,
 	requestedDuration time.Duration,
 	pollInterval int,
 	devInfoAnnouceInterval int,
@@ -243,6 +246,7 @@ func (runner *CollectorRunner) Run( //nolint:funlen // allow a slightly long fun
 	runner.initialise(
 		callback,
 		ptpInterface,
+		nodeName,
 		clientset,
 		pollInterval,
 		requestedDuration,