[POA-2609] Add k8s and CRI sub-modules and function interfaces (#66)

Child PRs: #67, #69, #71, #72, #73, #74, #75, #76, #77, #78, #79, #80, #82

Author: mudit-postman

apidump/apidump.go

@@ -2,6 +2,7 @@ package apidump
 
 import (
 	"context"
+	"net/http"
 	"os"
 	"os/exec"
 	"os/signal"
@@ -70,6 +71,14 @@ const (
 	notMatchedFilter filterState = "UNMATCHED"
 )
 
+// Args for running apidump as daemonset in Kubernetes
+type DaemonsetArgs struct {
+	TargetNetworkNamespaceOpt string
+	StopChan                  <-chan error `json:"-"`
+	APIKey                    string
+	Environment               string
+}
+
 type Args struct {
 	// Required args
 	ClientID akid.ClientID
@@ -138,6 +147,8 @@ type Args struct {
 
 	// Whether to enable repro mode and include request/response payloads when uploading witnesses.
 	ReproMode bool
+
+	DaemonsetArgs optionals.Optional[DaemonsetArgs]
 }
 
 // TODO: either remove write-to-local-HAR-file completely,
@@ -177,7 +188,14 @@ func (a *apidump) LookupService() error {
 	if !a.TargetIsRemote() {
 		return nil
 	}
-	frontClient := rest.NewFrontClient(a.Domain, a.ClientID)
+
+	// Set auth handler for processes starting via daemonset
+	var authHandler func(*http.Request) error
+	if daemonsetArgs, exists := a.DaemonsetArgs.Get(); exists {
+		authHandler = rest.ApiDumpDaemonsetAuthHandler(daemonsetArgs.APIKey, daemonsetArgs.Environment)
+	}
+
+	frontClient := rest.NewFrontClient(a.Domain, a.ClientID, authHandler)
 
 	if a.PostmanCollectionID != "" {
 		backendSvc, err := util.GetOrCreateServiceIDByPostmanCollectionID(frontClient, a.PostmanCollectionID)
@@ -197,7 +215,7 @@ func (a *apidump) LookupService() error {
 		a.backendSvcName = serviceName
 	}
 
-	a.learnClient = rest.NewLearnClient(a.Domain, a.ClientID, a.backendSvc)
+	a.learnClient = rest.NewLearnClient(a.Domain, a.ClientID, a.backendSvc, authHandler)
 	return nil
 }
 
@@ -398,7 +416,7 @@ func (a *apidump) RotateLearnSession(done <-chan struct{}, collectors []trace.Le
 
 		case <-t.C:
 			traceName := util.RandomLearnSessionName()
-			backendLrn, err := util.NewLearnSession(args.Domain, args.ClientID, a.backendSvc, traceName, traceTags, nil)
+			backendLrn, err := util.NewLearnSession(a.learnClient, traceName, traceTags, nil)
 			if err != nil {
 				telemetry.Error("new learn session", err)
 				printer.Errorf("Failed to create trace %s: %v\n", traceName, err)
@@ -494,8 +512,17 @@ func (a *apidump) Run() error {
 		printer.Debugln("Capturing filtered traffic for debugging.")
 	}
 
+	var (
+		targetNetworkNamespace   optionals.Optional[string]
+		daemonSetProcessStopChan <-chan error
+	)
+	if daemonsetArgs, exists := a.DaemonsetArgs.Get(); exists {
+		targetNetworkNamespace = optionals.Some(daemonsetArgs.TargetNetworkNamespaceOpt)
+		daemonSetProcessStopChan = daemonsetArgs.StopChan
+	}
+
 	// Get the interfaces to listen on.
-	interfaces, err := getEligibleInterfaces(args.Interfaces)
+	interfaces, err := getEligibleInterfaces(args.Interfaces, targetNetworkNamespace)
 	if err != nil {
 		a.SendErrorTelemetry(GetErrorTypeWithDefault(err, api_schema.ApidumpError_PCAPInterfaceOther), err)
 		return errors.Wrap(err, "No network interfaces could be used")
@@ -578,7 +605,7 @@ func (a *apidump) Run() error {
 	var backendLrn akid.LearnSessionID
 	if a.TargetIsRemote() {
 		uri := a.Out.AkitaURI
-		backendLrn, err = util.NewLearnSession(args.Domain, args.ClientID, a.backendSvc, uri.ObjectName, traceTags, nil)
+		backendLrn, err = util.NewLearnSession(a.learnClient, uri.ObjectName, traceTags, nil)
 		if err == nil {
 			printer.Infof("Created new trace on Postman Cloud: %s\n", uri)
 		} else {
@@ -767,7 +794,7 @@ func (a *apidump) Run() error {
 			go func(interfaceName, filter string) {
 				defer doneWG.Done()
 				// Collect trace. This blocks until stop is closed or an error occurs.
-				if err := pcap.Collect(stop, interfaceName, filter, bufferShare, args.ParseTLSHandshakes, collector, summary, pool); err != nil {
+				if err := pcap.Collect(stop, interfaceName, filter, targetNetworkNamespace, bufferShare, args.ParseTLSHandshakes, collector, summary, pool); err != nil {
 					errChan <- interfaceError{
 						interfaceName: interfaceName,
 						err:           errors.Wrapf(err, "failed to collect trace on interface %s", interfaceName),
@@ -884,6 +911,9 @@ func (a *apidump) Run() error {
 						printer.Stderr.Errorf("Encountered an error on interface %s.  Error: %s\n", interfaceErr.interfaceName, interfaceErr.err.Error())
 						break DoneWaitingForSignal
 					}
+				case externalStopError := <-daemonSetProcessStopChan:
+					printer.Stderr.Infof("Received external stop signal, error: %v\n", externalStopError)
+					break DoneWaitingForSignal
 				}
 			}
 		}

apidump/net.go

@@ -10,11 +10,12 @@ import (
 	"time"
 
 	"github.com/akitasoftware/akita-libs/api_schema"
-	"github.com/google/gopacket/pcap"
+	"github.com/akitasoftware/go-utils/optionals"
 	"github.com/pkg/errors"
 	"github.com/postmanlabs/postman-insights-agent/architecture"
 	"github.com/postmanlabs/postman-insights-agent/consts"
 	"github.com/postmanlabs/postman-insights-agent/env"
+	"github.com/postmanlabs/postman-insights-agent/pcap"
 	"github.com/postmanlabs/postman-insights-agent/printer"
 	"github.com/postmanlabs/postman-insights-agent/telemetry"
 )
@@ -96,7 +97,7 @@ func showPermissionErrors(sampleError error) error {
 // Get the list of interface names that we should listen on. By default, this is
 // all interfaces on the machine that are up. User may override this with
 // --interface flag.
-func getEligibleInterfaces(userSpecified []string) (map[string]interfaceInfo, error) {
+func getEligibleInterfaces(userSpecified []string, targetNetworkNamespaceOpt optionals.Optional[string]) (map[string]interfaceInfo, error) {
 	if len(userSpecified) > 0 {
 		results := make(map[string]interfaceInfo, len(userSpecified))
 		for _, n := range userSpecified {
@@ -107,7 +108,7 @@ func getEligibleInterfaces(userSpecified []string) (map[string]interfaceInfo, er
 			results[n] = iface
 		}
 
-		ifaceErrs := checkPcapPermissions(results)
+		ifaceErrs := checkPcapPermissions(results, targetNetworkNamespaceOpt)
 		for i, err := range ifaceErrs {
 			// Return error if we're not able to listen on a user-specified interface.
 			printer.Errorf("Error on interface %q: %v\n", i, err)
@@ -140,7 +141,7 @@ func getEligibleInterfaces(userSpecified []string) (map[string]interfaceInfo, er
 	// Don't return error if we're unable to listen to one of the available
 	// interfaces, and just listen to the interfaces we have the permissions
 	// for.
-	ifaceErrs := checkPcapPermissions(results)
+	ifaceErrs := checkPcapPermissions(results, targetNetworkNamespaceOpt)
 	var sampleError error
 	for ifaceName, err := range ifaceErrs {
 		printer.Warningf("Skipping interface %s for collecting packets because of error: %v\n", ifaceName, err)
@@ -184,7 +185,7 @@ func (pe pcapPermErr) Error() string {
 
 // Check if we have permission to capture packets on the given set of
 // interfaces.
-func checkPcapPermissions(interfaces map[string]interfaceInfo) map[string]error {
+func checkPcapPermissions(interfaces map[string]interfaceInfo, targetNetworkNamespaceOpt optionals.Optional[string]) map[string]error {
 	printer.Debugf("Checking pcap permissions...\n")
 	start := time.Now()
 
@@ -194,7 +195,7 @@ func checkPcapPermissions(interfaces map[string]interfaceInfo) map[string]error
 	for iface := range interfaces {
 		go func(iface string) {
 			defer wg.Done()
-			h, err := pcap.OpenLive(iface, 1600, true, pcap.BlockForever)
+			h, err := pcap.GetPcapHandle(iface, 1600, true, pcap.BlockForever, targetNetworkNamespaceOpt)
 			if err != nil {
 				telemetry.Error("pcap permissions", err)
 				errChan <- &pcapPermErr{iface: iface, err: err}

apispec/defaults.go

@@ -35,4 +35,7 @@ const (
 
 	// How often to rotate traces in the back end.
 	DefaultTraceRotateInterval = time.Hour
+
+	// Process all possible witness data
+	DefaultSampleRate = 1.0
 )

cmd/internal/apidump/apidump.go

@@ -256,7 +256,7 @@ func init() {
 	Cmd.Flags().Float64Var(
 		&sampleRateFlag,
 		"sample-rate",
-		1.0,
+		apispec.DefaultSampleRate,
 		"A number between [0.0, 1.0] to control sampling.",
 	)
 	Cmd.Flags().MarkDeprecated("sample-rate", "use --rate-limit instead.")

cmd/internal/cmderr/checks.go

@@ -42,7 +42,7 @@ func CheckAPIKeyAndInsightsProjectID(projectID string) error {
 		return errors.New("project ID is missing, it must be specified")
 	}
 
-	frontClient := rest.NewFrontClient(rest.Domain, telemetry.GetClientID())
+	frontClient := rest.NewFrontClient(rest.Domain, telemetry.GetClientID(), nil)
 	var serviceID akid.ServiceID
 	err = akid.ParseIDAs(projectID, &serviceID)
 	if err != nil {

cmd/internal/kube/daemonset/apidump_process.go

@@ -0,0 +1,116 @@
+package daemonset
+
+import (
+	"runtime/debug"
+
+	"github.com/akitasoftware/go-utils/optionals"
+	"github.com/pkg/errors"
+	"github.com/postmanlabs/postman-insights-agent/apidump"
+	"github.com/postmanlabs/postman-insights-agent/apispec"
+	"github.com/postmanlabs/postman-insights-agent/printer"
+	"github.com/postmanlabs/postman-insights-agent/rest"
+	"github.com/postmanlabs/postman-insights-agent/telemetry"
+	"k8s.io/apimachinery/pkg/types"
+)
+
+// StartApiDumpProcess initiates the API dump process for a given pod identified by its UID.
+// It retrieves the pod arguments, changes the pod's traffic monitoring state, and starts the API dump process in a separate goroutine.
+// The goroutine handles errors and state changes, and ensures the process is stopped properly.
+func (d *Daemonset) StartApiDumpProcess(podUID types.UID) error {
+	podArgs, err := d.getPodArgsFromMap(podUID)
+	if err != nil {
+		return err
+	}
+
+	err = podArgs.changePodTrafficMonitorState(TrafficMonitoringRunning, PodRunning)
+	if err != nil {
+		return errors.Wrapf(err, "failed to change pod state, pod name: %s, from: %s to: %s",
+			podArgs.PodName, podArgs.PodTrafficMonitorState, TrafficMonitoringRunning)
+	}
+
+	// Increment the wait group counter
+	d.ApidumpProcessesWG.Add(1)
+
+	go func() (funcErr error) {
+		// defer function handle the error (if any) in the apidump process and change the pod state accordingly
+		defer func() {
+			// Decrement the wait group counter
+			d.ApidumpProcessesWG.Done()
+
+			nextState := TrafficMonitoringEnded
+
+			if err := recover(); err != nil {
+				printer.Errorf("Panic occurred in apidump process for pod %s, err: %v\n%v\n",
+					podArgs.PodName, err, string(debug.Stack()))
+				nextState = TrafficMonitoringFailed
+			} else if funcErr != nil {
+				printer.Errorf("Error occurred in apidump process for pod %s, err: %v\n", podArgs.PodName, funcErr)
+				nextState = TrafficMonitoringFailed
+			} else {
+				printer.Infof("Apidump process ended for pod %s\n", podArgs.PodName)
+			}
+
+			// Move monitoring state to final apidump processing state
+			err = podArgs.changePodTrafficMonitorState(nextState,
+				TrafficMonitoringRunning, PodSucceeded, PodFailed, PodTerminated, DaemonSetShutdown)
+			if err != nil {
+				printer.Errorf("Failed to change pod state, pod name: %s, from: %s to: %s, error: %v\n",
+					podArgs.PodName, podArgs.PodTrafficMonitorState, nextState, err)
+				return
+			}
+		}()
+
+		networkNamespace, err := d.CRIClient.GetNetworkNamespace(podArgs.ContainerUUID)
+		if err != nil {
+			funcErr = errors.Errorf("Failed to get network namespace for pod/containerUUID: %s/%s, err: %v",
+				podArgs.PodName, podArgs.ContainerUUID, err)
+			return
+		}
+		// Prepend '/host' to network namespace, since '/proc' folder is mounted to '/host/proc'
+		networkNamespace = "/host" + networkNamespace
+
+		apidumpArgs := apidump.Args{
+			ClientID:                telemetry.GetClientID(),
+			Domain:                  rest.Domain,
+			ServiceID:               podArgs.InsightsProjectID,
+			SampleRate:              apispec.DefaultSampleRate,
+			WitnessesPerMinute:      apispec.DefaultRateLimit,
+			LearnSessionLifetime:    apispec.DefaultTraceRotateInterval,
+			TelemetryInterval:       apispec.DefaultTelemetryInterval_seconds,
+			ProcFSPollingInterval:   apispec.DefaultProcFSPollingInterval_seconds,
+			CollectTCPAndTLSReports: apispec.DefaultCollectTCPAndTLSReports,
+			ParseTLSHandshakes:      apispec.DefaultParseTLSHandshakes,
+			MaxWitnessSize_bytes:    apispec.DefaultMaxWitnessSize_bytes,
+			ReproMode:               d.InsightsReproModeEnabled,
+			DaemonsetArgs: optionals.Some(apidump.DaemonsetArgs{
+				TargetNetworkNamespaceOpt: networkNamespace,
+				StopChan:                  podArgs.StopChan,
+				APIKey:                    podArgs.PodCreds.InsightsAPIKey,
+				Environment:               podArgs.PodCreds.InsightsEnvironment,
+			}),
+		}
+
+		if err := apidump.Run(apidumpArgs); err != nil {
+			funcErr = errors.Wrapf(err, "failed to run apidump process for pod %s", podArgs.PodName)
+		}
+		return
+	}()
+
+	return nil
+}
+
+// StopApiDumpProcess signals the API dump process to stop for a given pod
+// identified by its UID. It retrieves the process's stop channel object from a map
+// and sends a stop signal to trigger apidump shutdown.
+func (d *Daemonset) StopApiDumpProcess(podUID types.UID, stopErr error) error {
+	podArgs, err := d.getPodArgsFromMap(podUID)
+	if err != nil {
+		return err
+	}
+
+	printer.Infof("Stopping API dump process for pod %s\n", podArgs.PodName)
+	podArgs.StopChan <- stopErr
+	close(podArgs.StopChan)
+
+	return nil
+}

cmd/internal/kube/daemonset/constants.go

@@ -0,0 +1,19 @@
+package daemonset
+
+import "time"
+
+const (
+	// Pod environment variables
+	POSTMAN_INSIGHTS_PROJECT_ID = "POSTMAN_INSIGHTS_PROJECT_ID"
+	POSTMAN_INSIGHTS_API_KEY    = "POSTMAN_INSIGHTS_API_KEY"
+
+	// Daemonset environment variables
+	POSTMAN_INSIGHTS_ENV                = "POSTMAN_ENV" // This is same as root POSTMAN_ENV
+	POSTMAN_INSIGHTS_VERIFICATION_TOKEN = "POSTMAN_INSIGHTS_VERIFICATION_TOKEN"
+	POSTMAN_INSIGHTS_CLUSTER_NAME       = "POSTMAN_INSIGHTS_CLUSTER_NAME"
+	POSTMAN_INSIGHTS_REPRO_MODE_ENABLED = "POSTMAN_INSIGHTS_REPRO_MODE_ENABLED"
+
+	// Workers intervals
+	DefaultTelemetryInterval      = 5 * time.Minute
+	DefaultPodHealthCheckInterval = 5 * time.Minute
+)