Fix: [EKS] server could not find the requested resourceevents.events.k8s.io (#565)

agent/pkg/utils/utils.go

@@ -48,7 +48,8 @@ func StartServer(app *gin.Engine) {
 func GetTappedPodsStatus() []shared.TappedPodStatus {
 	tappedPodsStatus := make([]shared.TappedPodStatus, 0)
 	for _, pod := range providers.TapStatus.Pods {
-		isTapped := strings.ToLower(providers.TappersStatus[pod.NodeName].Status) == "started"
+		status := strings.ToLower(providers.TappersStatus[pod.NodeName].Status)
+		isTapped := status == "running"
 		tappedPodsStatus = append(tappedPodsStatus, shared.TappedPodStatus{Name: pod.Name, Namespace: pod.Namespace, IsTapped: isTapped})
 	}
 	return tappedPodsStatus

cli/cmd/common.go

@@ -2,6 +2,7 @@ package cmd
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"github.com/up9inc/mizu/cli/apiserver"
@@ -9,6 +10,7 @@ import (
 	"github.com/up9inc/mizu/cli/mizu/fsUtils"
 	"github.com/up9inc/mizu/cli/resources"
 	"github.com/up9inc/mizu/cli/telemetry"
+	"github.com/up9inc/mizu/shared"
 	"path"
 	"time"
 
@@ -71,3 +73,12 @@ func dumpLogsIfNeeded(ctx context.Context, kubernetesProvider *kubernetes.Provid
 		logger.Log.Errorf("Failed dump logs %v", err)
 	}
 }
+
+func getSerializedMizuAgentConfig(mizuAgentConfig *shared.MizuAgentConfig) (string, error) {
+	serializedConfig, err := json.Marshal(mizuAgentConfig)
+	if err != nil {
+		return "", err
+	}
+
+	return string(serializedConfig), nil
+}

cli/cmd/tapRunner.go

@@ -33,8 +33,8 @@ import (
 const cleanupTimeout = time.Minute
 
 type tapState struct {
-	startTime        time.Time
-	targetNamespaces []string
+	startTime                time.Time
+	targetNamespaces         []string
 	mizuServiceAccountExists bool
 }
 
@@ -94,9 +94,15 @@ func RunMizuTap() {
 
 	state.targetNamespaces = getNamespaces(kubernetesProvider)
 
-	serializedMizuConfig, err := config.GetSerializedMizuAgentConfig(state.targetNamespaces, mizuApiFilteringOptions)
+	mizuAgentConfig, err := getMizuAgentConfig(state.targetNamespaces, mizuApiFilteringOptions)
 	if err != nil {
-		logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Error composing mizu config: %v", errormessage.FormatError(err)))
+		logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Error getting mizu config: %v", errormessage.FormatError(err)))
+		return
+	}
+
+	serializedMizuConfig, err := getSerializedMizuAgentConfig(mizuAgentConfig)
+	if err != nil {
+		logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Error serializing mizu config: %v", errormessage.FormatError(err)))
 		return
 	}
 
@@ -165,6 +171,31 @@ func handleDaemonModePostCreation(cancel context.CancelFunc, kubernetesProvider
 	return nil
 }
 
+func getMizuAgentConfig(targetNamespaces []string, mizuApiFilteringOptions *api.TrafficFilteringOptions) (*shared.MizuAgentConfig, error) {
+	serializableRegex, err := api.CompileRegexToSerializableRegexp(config.Config.Tap.PodRegexStr)
+	if err != nil {
+		return nil, err
+	}
+
+	mizuAgentConfig := shared.MizuAgentConfig{
+		TapTargetRegex:          *serializableRegex,
+		MaxDBSizeBytes:          config.Config.Tap.MaxEntriesDBSizeBytes(),
+		TargetNamespaces:        targetNamespaces,
+		AgentImage:              config.Config.AgentImage,
+		PullPolicy:              config.Config.ImagePullPolicyStr,
+		LogLevel:                config.Config.LogLevel(),
+		IgnoredUserAgents:       config.Config.Tap.IgnoredUserAgents,
+		TapperResources:         config.Config.Tap.TapperResources,
+		MizuResourcesNamespace:  config.Config.MizuResourcesNamespace,
+		MizuApiFilteringOptions: *mizuApiFilteringOptions,
+		AgentDatabasePath:       shared.DataDirPath,
+		Istio:                   config.Config.Tap.Istio,
+		SyncTappers:             config.Config.Tap.DaemonMode,
+	}
+
+	return &mizuAgentConfig, nil
+}
+
 /*
 this function is a bit problematic as it might be detached from the actual pods the mizu api server will tap.
 The alternative would be to wait for api server to be ready and then query it for the pods it listens to, this has
@@ -389,7 +420,8 @@ func watchApiServerEvents(ctx context.Context, kubernetesProvider *kubernetes.Pr
 
 			event, err := wEvent.ToEvent()
 			if err != nil {
-				logger.Log.Errorf(fmt.Sprintf("Error parsing Mizu resource event: %+v", err))
+				logger.Log.Debugf("[ERROR] parsing Mizu resource event: %+v", err)
+				continue
 			}
 
 			if state.startTime.After(event.CreationTimestamp.Time) {
@@ -417,7 +449,7 @@ func watchApiServerEvents(ctx context.Context, kubernetesProvider *kubernetes.Pr
 				continue
 			}
 
-			logger.Log.Errorf("Watching API server events loop, error: %+v", err)
+			logger.Log.Debugf("[Error] Watching API server events loop, error: %+v", err)
 		case <-ctx.Done():
 			logger.Log.Debugf("Watching API server events loop, ctx done")
 			return

cli/config/config.go

@@ -9,9 +9,6 @@ import (
 	"strconv"
 	"strings"
 
-	"github.com/up9inc/mizu/tap/api"
-	"k8s.io/apimachinery/pkg/util/json"
-
 	"github.com/up9inc/mizu/shared"
 	"github.com/up9inc/mizu/shared/logger"
 
@@ -371,38 +368,3 @@ func setZeroForReadonlyFields(currentElem reflect.Value) {
 		}
 	}
 }
-
-func GetSerializedMizuAgentConfig(targetNamespaces []string, mizuApiFilteringOptions *api.TrafficFilteringOptions) (string, error) {
-	mizuConfig, err := getMizuAgentConfig(targetNamespaces, mizuApiFilteringOptions)
-	if err != nil {
-		return "", err
-	}
-	serializedConfig, err := json.Marshal(mizuConfig)
-	if err != nil {
-		return "", err
-	}
-	return string(serializedConfig), nil
-}
-
-func getMizuAgentConfig(targetNamespaces []string, mizuApiFilteringOptions *api.TrafficFilteringOptions) (*shared.MizuAgentConfig, error) {
-	serializableRegex, err := api.CompileRegexToSerializableRegexp(Config.Tap.PodRegexStr)
-	if err != nil {
-		return nil, err
-	}
-	config := shared.MizuAgentConfig{
-		TapTargetRegex:          *serializableRegex,
-		MaxDBSizeBytes:          Config.Tap.MaxEntriesDBSizeBytes(),
-		TargetNamespaces:        targetNamespaces,
-		AgentImage:              Config.AgentImage,
-		PullPolicy:              Config.ImagePullPolicyStr,
-		LogLevel:                Config.LogLevel(),
-		IgnoredUserAgents:       Config.Tap.IgnoredUserAgents,
-		TapperResources:         Config.Tap.TapperResources,
-		MizuResourcesNamespace:  Config.MizuResourcesNamespace,
-		MizuApiFilteringOptions: *mizuApiFilteringOptions,
-		AgentDatabasePath:       shared.DataDirPath,
-		Istio:                   Config.Tap.Istio,
-		SyncTappers:             Config.Tap.DaemonMode,
-	}
-	return &config, nil
-}

shared/kubernetes/mizuTapperSyncer.go

@@ -70,9 +70,49 @@ func CreateAndStartMizuTapperSyncer(ctx context.Context, kubernetesProvider *Pro
 
 	go syncer.watchPodsForTapping()
 	go syncer.watchTapperEvents()
+	go syncer.watchTapperPods()
 	return syncer, nil
 }
 
+func (tapperSyncer *MizuTapperSyncer) watchTapperPods() {
+	mizuResourceRegex := regexp.MustCompile(fmt.Sprintf("^%s.*", TapperPodName))
+	podWatchHelper := NewPodWatchHelper(tapperSyncer.kubernetesProvider, mizuResourceRegex)
+	eventChan, errorChan := FilteredWatch(tapperSyncer.context, podWatchHelper, []string{tapperSyncer.config.MizuResourcesNamespace}, podWatchHelper)
+
+	for {
+		select {
+		case wEvent, ok := <-eventChan:
+			if !ok {
+				eventChan = nil
+				continue
+			}
+
+			pod, err := wEvent.ToPod()
+			if err != nil {
+				logger.Log.Debugf("[ERROR] parsing Mizu resource pod: %+v", err)
+				continue
+			}
+
+			logger.Log.Debugf("Watching tapper pods loop, tapper: %v, node: %v, status: %v", pod.Name, pod.Spec.NodeName, pod.Status.Phase)
+			if pod.Spec.NodeName != "" {
+				tapperStatus := shared.TapperStatus{TapperName: pod.Name, NodeName: pod.Spec.NodeName, Status: string(pod.Status.Phase)}
+				tapperSyncer.TapperStatusChangedOut <- tapperStatus
+			}
+
+		case err, ok := <-errorChan:
+			if !ok {
+				errorChan = nil
+				continue
+			}
+			logger.Log.Debugf("[ERROR] Watching tapper pods loop, error: %+v", err)
+
+		case <-tapperSyncer.context.Done():
+			logger.Log.Debugf("Watching tapper pods loop, ctx done")
+			return
+		}
+	}
+}
+
 func (tapperSyncer *MizuTapperSyncer) watchTapperEvents() {
 	mizuResourceRegex := regexp.MustCompile(fmt.Sprintf("^%s.*", TapperPodName))
 	eventWatchHelper := NewEventWatchHelper(tapperSyncer.kubernetesProvider, mizuResourceRegex, "pod")
@@ -88,7 +128,8 @@ func (tapperSyncer *MizuTapperSyncer) watchTapperEvents() {
 
 			event, err := wEvent.ToEvent()
 			if err != nil {
-				logger.Log.Errorf(fmt.Sprintf("Error parsing Mizu resource event: %+v", err))
+				logger.Log.Debugf("[ERROR] parsing Mizu resource event: %+v", err)
+				continue
 			}
 
 			if tapperSyncer.startTime.After(event.CreationTimestamp.Time) {
@@ -117,8 +158,8 @@ func (tapperSyncer *MizuTapperSyncer) watchTapperEvents() {
 				nodeName = pod.Spec.Affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0].MatchFields[0].Values[0]
 			}
 
-			taperStatus := shared.TapperStatus{TapperName: pod.Name, NodeName: nodeName, Status: event.Reason}
-			tapperSyncer.TapperStatusChangedOut <- taperStatus
+			tapperStatus := shared.TapperStatus{TapperName: pod.Name, NodeName: nodeName, Status: string(pod.Status.Phase)}
+			tapperSyncer.TapperStatusChangedOut <- tapperStatus
 
 		case err, ok := <-errorChan:
 			if !ok {
@@ -126,7 +167,7 @@ func (tapperSyncer *MizuTapperSyncer) watchTapperEvents() {
 				continue
 			}
 
-			logger.Log.Errorf("Watching tapper events loop, error: %+v", err)
+			logger.Log.Debugf("[ERROR] Watching tapper events loop, error: %+v", err)
 
 		case <-tapperSyncer.context.Done():
 			logger.Log.Debugf("Watching tapper events loop, ctx done")