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

* Update main.go and passive_tapper.go

* Update main.go

acceptanceTests/go.sum

@@ -110,6 +110,7 @@ github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9
 github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
 github.com/go-logr/logr v0.1.0/go.mod h1:ixOQHD9gLJUVQQ2ZOR7zLEifBX6tGkNJF4QyIY7sIas=
 github.com/go-logr/logr v0.2.0/go.mod h1:z6/tIYblkpsD+a4lm/fGIIU9mZ+XfAiaFtq7xTgseGU=
+github.com/go-logr/logr v0.4.0 h1:K7/B1jt6fIBQVd4Owv2MqGQClcgf0R266+7C/QjRcLc=
 github.com/go-logr/logr v0.4.0/go.mod h1:z6/tIYblkpsD+a4lm/fGIIU9mZ+XfAiaFtq7xTgseGU=
 github.com/go-openapi/analysis v0.0.0-20180825180245-b006789cd277/go.mod h1:k70tL6pCuVxPJOHXQ+wIac1FUrvNkHolPie/cLEU6hI=
 github.com/go-openapi/analysis v0.17.0/go.mod h1:IowGgpVeD0vNm45So8nr+IcQ3pxVtpRoBWb8PVZO0ik=
@@ -156,6 +157,7 @@ github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/me
 github.com/gobuffalo/here v0.6.0/go.mod h1:wAG085dHOYqUpf+Ap+WOdrPTp5IYcDAs/x7PLa8Y5fM=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
 github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4=
+github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
 github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
 github.com/golang-jwt/jwt/v4 v4.1.0 h1:XUgk2Ex5veyVFVeLm0xhusUTQybEbexJXrvPNOKkSY0=
 github.com/golang-jwt/jwt/v4 v4.1.0/go.mod h1:/xlHOz8bRuivTWchD4jCa+NbatV+wEUSzwAxVc6locg=
@@ -191,8 +193,10 @@ github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMyw
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.4 h1:L8R9j+yAqZuZjsqh/z+F1NCffTKKLShY6zXTItVIZ8M=
 github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
+github.com/google/gofuzz v1.1.0 h1:Hsa8mG0dQ46ij8Sl2AYJDUv1oA9/d6Vk+3LG99Oe02g=
 github.com/google/gofuzz v1.1.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/martian v2.1.0+incompatible h1:/CP5g8u/VJHijgedC/Legn3BAbAaWPgecwXBIDzw5no=
 github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
@@ -243,6 +247,7 @@ github.com/imdario/mergo v0.3.5/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJ
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo=
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
+github.com/json-iterator/go v1.1.10 h1:Kz6Cvnvv2wGdaG/V8yMvfkmNiXq9Ya2KUv4rouJJr68=
 github.com/json-iterator/go v1.1.10/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
 github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod h1:6v2b51hI/fHJwM22ozAgKL4VKDeJcHhJFhtBdhmNjmU=
 github.com/jstemmer/go-junit-report v0.9.1/go.mod h1:Brl9GWCQeLvo8nXZwPNNblvFj/XSXhF0NWZEnDohbsk=
@@ -288,8 +293,10 @@ github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh
 github.com/moby/spdystream v0.2.0/go.mod h1:f7i0iNDQJ059oMTcWxx8MA/zKFIuD/lY+0GqbN2Wy8c=
 github.com/moby/term v0.0.0-20201216013528-df9cb8a40635/go.mod h1:FBS0z0QWA44HXygs7VXDUOGoN/1TV3RuWkLO04am3wc=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
+github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
+github.com/modern-go/reflect2 v1.0.1 h1:9f412s+6RmYXLWZSEzVVgPGK7C2PphHj5RJrvfx9AWI=
 github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
 github.com/monochromegane/go-gitignore v0.0.0-20200626010858-205db1a8cc00/go.mod h1:Pm3mSP3c5uWn86xMLZ5Sa7JB9GsEZySvHYXCTK4E9q4=
 github.com/munnerz/goautoneg v0.0.0-20120707110453-a547fc61f48d/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
@@ -462,6 +469,7 @@ golang.org/x/net v0.0.0-20200301022130-244492dfa37a/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20200324143707-d3edc9973b7e/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20201110031124-69a78807bb2b/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
+golang.org/x/net v0.0.0-20210224082022-3d97a244fca7 h1:OgUuv8lsRpBibGNbSizVwKWlysjaNzmC9gYMhPVfqFM=
 golang.org/x/net v0.0.0-20210224082022-3d97a244fca7/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -521,6 +529,7 @@ golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.4 h1:0YWbFKbhXG/wIiuHDSKpS0Iy7FSA+u45VtBMfQcFTTc=
 golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
@@ -630,6 +639,7 @@ gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f h1:BLraFXnmrev5lT+xlilqcH8X
 gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
+gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc=
 gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw=
 gopkg.in/ini.v1 v1.51.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
 gopkg.in/resty.v1 v1.12.0/go.mod h1:mDo4pnntr5jdWRML875a/NmxYqAlA73dVijT2AXvQQo=
@@ -641,6 +651,7 @@ gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.5/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.7/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
 gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b h1:h8qDotaEPuJATrMmW04NCwg7v22aHH28wwpauUhK9Oo=
@@ -653,7 +664,9 @@ honnef.co/go/tools v0.0.0-20190418001031-e561f6794a2a/go.mod h1:rf3lG4BRIbNafJWh
 honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.1-2019.2.3/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt0JzvZhAg=
 honnef.co/go/tools v0.0.1-2020.1.3/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
+k8s.io/api v0.21.2 h1:vz7DqmRsXTCSa6pNxXwQ1IYeAZgdIsua+DZU+o+SX3Y=
 k8s.io/api v0.21.2/go.mod h1:Lv6UGJZ1rlMI1qusN8ruAp9PUBFyBwpEHAdG24vIsiU=
+k8s.io/apimachinery v0.21.2 h1:vezUc/BHqWlQDnZ+XkrpXSmnANSLbpnlpwo0Lhk0gpc=
 k8s.io/apimachinery v0.21.2/go.mod h1:CdTY8fU/BlvAbJ2z/8kBwimGki5Zp8/fbVuLY8gJumM=
 k8s.io/cli-runtime v0.21.2/go.mod h1:8u/jFcM0QpoI28f6sfrAAIslLCXUYKD5SsPPMWiHYrI=
 k8s.io/client-go v0.21.2/go.mod h1:HdJ9iknWpbl3vMGtib6T2PyI/VYxiZfq936WNVHBRrA=
@@ -664,6 +677,7 @@ k8s.io/gengo v0.0.0-20200413195148-3a45101e95ac/go.mod h1:ezvh/TsK7cY6rbqRK0oQQ8
 k8s.io/gengo v0.0.0-20201214224949-b6c5ce23f027/go.mod h1:FiNAH4ZV3gBg2Kwh89tzAEV2be7d5xI0vBa/VySYy3E=
 k8s.io/klog/v2 v2.0.0/go.mod h1:PBfzABfn139FHAV07az/IF9Wp1bkk3vpT2XSJ76fSDE=
 k8s.io/klog/v2 v2.2.0/go.mod h1:Od+F08eJP+W3HUb4pSrPpgp9DGU4GzlpG/TmITuYh/Y=
+k8s.io/klog/v2 v2.8.0 h1:Q3gmuM9hKEjefWFFYF0Mat+YyFJvsUyYuwyNNJ5C9Ts=
 k8s.io/klog/v2 v2.8.0/go.mod h1:hy9LJ/NvuK+iVyP4Ehqva4HxZG/oXyIS3n3Jmire4Ec=
 k8s.io/kube-openapi v0.0.0-20210305001622-591a79e4bda7/go.mod h1:wXW5VT87nVfh/iLV8FpR2uDvrFyomxbtb1KivDbvPTE=
 k8s.io/kubectl v0.21.2/go.mod h1:PgeUclpG8VVmmQIl8zpLar3IQEpFc9mrmvlwY3CK1xo=
@@ -677,5 +691,6 @@ sigs.k8s.io/kustomize/cmd/config v0.9.10/go.mod h1:Mrby0WnRH7hA6OwOYnYpfpiY0WJIM
 sigs.k8s.io/kustomize/kustomize/v4 v4.1.2/go.mod h1:PxBvo4WGYlCLeRPL+ziT64wBXqbgfcalOS/SXa/tcyo=
 sigs.k8s.io/kustomize/kyaml v0.10.17/go.mod h1:mlQFagmkm1P+W4lZJbJ/yaxMd8PqMRSC4cPcfUVt5Hg=
 sigs.k8s.io/structured-merge-diff/v4 v4.0.2/go.mod h1:bJZC9H9iH24zzfZ/41RGcq60oK1F7G282QMXDPYydCw=
+sigs.k8s.io/structured-merge-diff/v4 v4.1.0 h1:C4r9BgJ98vrKnnVCjwCSXcWjWe0NKcUQkmzDXZXGwH8=
 sigs.k8s.io/structured-merge-diff/v4 v4.1.0/go.mod h1:bJZC9H9iH24zzfZ/41RGcq60oK1F7G282QMXDPYydCw=
 sigs.k8s.io/yaml v1.2.0/go.mod h1:yfXDCHCao9+ENCvLSE62v9VSji2MKu5jeNfTrofGhJc=

agent/main.go

@@ -55,7 +55,7 @@ var extensionsMap map[string]*tapApi.Extension // global
 var startTime int64
 
 const (
-	socketConnectionRetries    = 10
+	socketConnectionRetries    = 30
 	socketConnectionRetryDelay = time.Second * 2
 	socketHandshakeTimeout     = time.Second * 2
 )
@@ -131,6 +131,10 @@ func main() {
 			}
 		}
 
+		if config.Config.SyncTappers {
+			startSyncingTappers()
+		}
+
 		hostApi(outputItemsChannel)
 	} else if *harsReaderMode {
 		outputItemsChannel := make(chan *tapApi.OutputChannelItem, 1000)
@@ -260,20 +264,6 @@ func hostApi(socketHarOutputChannel chan<- *tapApi.OutputChannelItem) {
 	routes.StatusRoutes(app)
 	routes.NotFoundRoute(app)
 
-	if config.Config.DaemonMode {
-		ctx, cancel := context.WithCancel(context.Background())
-		defer cancel()
-
-		kubernetesProvider, err := kubernetes.NewProviderInCluster()
-		if err != nil {
-			logger.Log.Fatalf("error creating k8s provider: %+v", err)
-		}
-
-		if _, err := startMizuTapperSyncer(ctx, kubernetesProvider); err != nil {
-			logger.Log.Fatalf("error initializing tapper syncer: %+v", err)
-		}
-	}
-
 	utils.StartServer(app)
 }
 
@@ -425,12 +415,56 @@ func dialSocketWithRetry(socketAddress string, retryAmount int, retryDelay time.
 				time.Sleep(retryDelay)
 			}
 		} else {
+			go handleIncomingMessageAsTapper(socketConnection)
 			return socketConnection, nil
 		}
 	}
 	return nil, lastErr
 }
 
+func handleIncomingMessageAsTapper(socketConnection *websocket.Conn) {
+	for {
+		if _, message, err := socketConnection.ReadMessage(); err != nil {
+			logger.Log.Errorf("error reading message from socket connection, err: %s, (%v,%+v)", err, err, err)
+			if errors.Is(err, syscall.EPIPE) {
+				// socket has disconnected, we can safely stop this goroutine
+				return
+			}
+		} else {
+			var socketMessageBase shared.WebSocketMessageMetadata
+			if err := json.Unmarshal(message, &socketMessageBase); err != nil {
+				logger.Log.Errorf("Could not unmarshal websocket message %v", err)
+			} else {
+				switch socketMessageBase.MessageType {
+				case shared.WebSocketMessageTypeTapConfig:
+					var tapConfigMessage *shared.WebSocketTapConfigMessage
+					if err := json.Unmarshal(message, &tapConfigMessage); err != nil {
+						logger.Log.Errorf("received unknown message from socket connection: %s, err: %s, (%v,%+v)", string(message), err, err, err)
+					} else {
+						tap.UpdateTapTargets(tapConfigMessage.TapTargets)
+					}
+				default:
+					logger.Log.Warningf("Received socket message of type %s for which no handlers are defined", socketMessageBase.MessageType)
+				}
+			}
+		}
+	}
+}
+
+func startSyncingTappers() {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	kubernetesProvider, err := kubernetes.NewProviderInCluster()
+	if err != nil {
+		logger.Log.Fatalf("error creating k8s provider: %+v", err)
+	}
+
+	if _, err := startMizuTapperSyncer(ctx, kubernetesProvider); err != nil {
+		logger.Log.Fatalf("error initializing tapper syncer: %+v", err)
+	}
+}
+
 func startMizuTapperSyncer(ctx context.Context, provider *kubernetes.Provider) (*kubernetes.MizuTapperSyncer, error) {
 	tapperSyncer, err := kubernetes.CreateAndStartMizuTapperSyncer(ctx, provider, kubernetes.TapperSyncerConfig{
 		TargetNamespaces:         config.Config.TargetNamespaces,

agent/pkg/config/config.go

@@ -56,6 +56,6 @@ func getDefaultConfig() (*shared.MizuAgentConfig, error) {
 		TapTargetRegex:    *regex,
 		MaxDBSizeBytes:    defaultMaxDatabaseSizeBytes,
 		AgentDatabasePath: DefaultDatabasePath,
-		DaemonMode:        false,
+		SyncTappers:       false,
 	}, nil
 }

agent/pkg/controllers/status_controller.go

@@ -17,7 +17,7 @@ import (
 )
 
 func HealthCheck(c *gin.Context) {
-	if config.Config.DaemonMode {
+	if config.Config.SyncTappers {
 		if providers.ExpectedTapperAmount != providers.TappersCount {
 			c.JSON(http.StatusInternalServerError, fmt.Sprintf("expecting more tappers than are actually connected (%d expected, %d connected)", providers.ExpectedTapperAmount, providers.TappersCount))
 			return

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/cleanRunner.go

@@ -1,6 +1,9 @@
 package cmd
 
-import "github.com/up9inc/mizu/cli/apiserver"
+import (
+	"github.com/up9inc/mizu/cli/apiserver"
+	"github.com/up9inc/mizu/cli/config"
+)
 
 func performCleanCommand() {
 	kubernetesProvider, err := getKubernetesProviderForCli()
@@ -8,5 +11,5 @@ func performCleanCommand() {
 		return
 	}
 
-	finishMizuExecution(kubernetesProvider, apiserver.NewProvider(GetApiServerUrl(), apiserver.DefaultRetries, apiserver.DefaultTimeout))
+	finishMizuExecution(kubernetesProvider, apiserver.NewProvider(GetApiServerUrl(), apiserver.DefaultRetries, apiserver.DefaultTimeout), config.Config.IsNsRestrictedMode(), config.Config.MizuResourcesNamespace)
 }

cli/cmd/common.go

@@ -2,16 +2,16 @@ package cmd
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"github.com/up9inc/mizu/cli/apiserver"
 	"github.com/up9inc/mizu/cli/mizu"
 	"github.com/up9inc/mizu/cli/mizu/fsUtils"
+	"github.com/up9inc/mizu/cli/resources"
 	"github.com/up9inc/mizu/cli/telemetry"
-	"os"
-	"os/signal"
+	"github.com/up9inc/mizu/shared"
 	"path"
-	"syscall"
 	"time"
 
 	"github.com/up9inc/mizu/cli/config"
@@ -37,22 +37,6 @@ func startProxyReportErrorIfAny(kubernetesProvider *kubernetes.Provider, cancel
 	logger.Log.Debugf("proxy ended")
 }
 
-func waitForFinish(ctx context.Context, cancel context.CancelFunc) {
-	logger.Log.Debugf("waiting for finish...")
-	sigChan := make(chan os.Signal, 1)
-	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
-
-	// block until ctx cancel is called or termination signal is received
-	select {
-	case <-ctx.Done():
-		logger.Log.Debugf("ctx done")
-		break
-	case <-sigChan:
-		logger.Log.Debugf("Got termination signal, canceling execution...")
-		cancel()
-	}
-}
-
 func getKubernetesProviderForCli() (*kubernetes.Provider, error) {
 	kubernetesProvider, err := kubernetes.NewProvider(config.Config.KubeConfigPath())
 	if err != nil {
@@ -71,12 +55,12 @@ func handleKubernetesProviderError(err error) {
 	}
 }
 
-func finishMizuExecution(kubernetesProvider *kubernetes.Provider, apiProvider *apiserver.Provider) {
+func finishMizuExecution(kubernetesProvider *kubernetes.Provider, apiProvider *apiserver.Provider, isNsRestrictedMode bool, mizuResourcesNamespace string) {
 	telemetry.ReportAPICalls(apiProvider)
 	removalCtx, cancel := context.WithTimeout(context.Background(), cleanupTimeout)
 	defer cancel()
 	dumpLogsIfNeeded(removalCtx, kubernetesProvider)
-	cleanUpMizuResources(removalCtx, cancel, kubernetesProvider)
+	resources.CleanUpMizuResources(removalCtx, cancel, kubernetesProvider, isNsRestrictedMode, mizuResourcesNamespace)
 }
 
 func dumpLogsIfNeeded(ctx context.Context, kubernetesProvider *kubernetes.Provider) {
@@ -90,22 +74,11 @@ func dumpLogsIfNeeded(ctx context.Context, kubernetesProvider *kubernetes.Provid
 	}
 }
 
-func cleanUpMizuResources(ctx context.Context, cancel context.CancelFunc, kubernetesProvider *kubernetes.Provider) {
-	logger.Log.Infof("\nRemoving mizu resources")
-
-	var leftoverResources []string
-
-	if config.Config.IsNsRestrictedMode() {
-		leftoverResources = cleanUpRestrictedMode(ctx, kubernetesProvider)
-	} else {
-		leftoverResources = cleanUpNonRestrictedMode(ctx, cancel, kubernetesProvider)
+func getSerializedMizuAgentConfig(mizuAgentConfig *shared.MizuAgentConfig) (string, error) {
+	serializedConfig, err := json.Marshal(mizuAgentConfig)
+	if err != nil {
+		return "", err
 	}
 
-	if len(leftoverResources) > 0 {
-		errMsg := fmt.Sprintf("Failed to remove the following resources, for more info check logs at %s:", fsUtils.GetLogFilePath())
-		for _, resource := range leftoverResources {
-			errMsg += "\n- " + resource
-		}
-		logger.Log.Errorf(uiUtils.Error, errMsg)
-	}
+	return string(serializedConfig), nil
 }

cli/cmd/tapRunner.go

@@ -4,26 +4,24 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/up9inc/mizu/cli/resources"
+	"github.com/up9inc/mizu/cli/utils"
 	"io/ioutil"
 	"regexp"
 	"strings"
 	"time"
 
-	"k8s.io/apimachinery/pkg/util/intstr"
-
 	"github.com/getkin/kin-openapi/openapi3"
 	"gopkg.in/yaml.v3"
 	core "k8s.io/api/core/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/util/wait"
 
 	"github.com/up9inc/mizu/cli/apiserver"
 	"github.com/up9inc/mizu/cli/cmd/goUtils"
 	"github.com/up9inc/mizu/cli/config"
 	"github.com/up9inc/mizu/cli/config/configStructs"
 	"github.com/up9inc/mizu/cli/errormessage"
-	"github.com/up9inc/mizu/cli/mizu"
 	"github.com/up9inc/mizu/cli/mizu/fsUtils"
 	"github.com/up9inc/mizu/cli/uiUtils"
 	"github.com/up9inc/mizu/shared"
@@ -35,11 +33,8 @@ import (
 const cleanupTimeout = time.Minute
 
 type tapState struct {
-	startTime        time.Time
-	targetNamespaces []string
-
-	apiServerService         *core.Service
-	tapperSyncer             *kubernetes.MizuTapperSyncer
+	startTime                time.Time
+	targetNamespaces         []string
 	mizuServiceAccountExists bool
 }
 
@@ -99,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
 	}
 
@@ -131,7 +132,7 @@ func RunMizuTap() {
 	}
 
 	logger.Log.Infof("Waiting for Mizu Agent to start...")
-	if err := createMizuResources(ctx, cancel, kubernetesProvider, serializedValidationRules, serializedContract, serializedMizuConfig); err != nil {
+	if state.mizuServiceAccountExists, err = resources.CreateMizuResources(ctx, cancel, kubernetesProvider, serializedValidationRules, serializedContract, serializedMizuConfig, config.Config.IsNsRestrictedMode(), config.Config.MizuResourcesNamespace, config.Config.Tap.DaemonMode, config.Config.AgentImage, getSyncEntriesConfig(), config.Config.Tap.MaxEntriesDBSizeBytes(), config.Config.Tap.ApiServerResources, config.Config.ImagePullPolicy(), config.Config.LogLevel(), config.Config.Tap.NoPersistentVolumeClaim); err != nil {
 		logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Error creating resources: %v", errormessage.FormatError(err)))
 
 		var statusError *k8serrors.StatusError
@@ -143,24 +144,24 @@ func RunMizuTap() {
 		return
 	}
 	if config.Config.Tap.DaemonMode {
-		if err := handleDaemonModePostCreation(ctx, cancel, kubernetesProvider, state.targetNamespaces); err != nil {
-			defer finishMizuExecution(kubernetesProvider, apiProvider)
+		if err := handleDaemonModePostCreation(cancel, kubernetesProvider); err != nil {
+			defer finishMizuExecution(kubernetesProvider, apiProvider, config.Config.IsNsRestrictedMode(), config.Config.MizuResourcesNamespace)
 			cancel()
 		} else {
 			logger.Log.Infof(uiUtils.Magenta, "Mizu is now running in daemon mode, run `mizu view` to connect to the mizu daemon instance")
 		}
 	} else {
-		defer finishMizuExecution(kubernetesProvider, apiProvider)
+		defer finishMizuExecution(kubernetesProvider, apiProvider, config.Config.IsNsRestrictedMode(), config.Config.MizuResourcesNamespace)
 
 		go goUtils.HandleExcWrapper(watchApiServerEvents, ctx, kubernetesProvider, cancel)
 		go goUtils.HandleExcWrapper(watchApiServerPod, ctx, kubernetesProvider, cancel)
 
 		// block until exit signal or error
-		waitForFinish(ctx, cancel)
+		utils.WaitForFinish(ctx, cancel)
 	}
 }
 
-func handleDaemonModePostCreation(ctx context.Context, cancel context.CancelFunc, kubernetesProvider *kubernetes.Provider, namespaces []string) error {
+func handleDaemonModePostCreation(cancel context.CancelFunc, kubernetesProvider *kubernetes.Provider) error {
 	apiProvider := apiserver.NewProvider(GetApiServerUrl(), 90, 1*time.Second)
 
 	if err := waitForDaemonModeToBeReady(cancel, kubernetesProvider, apiProvider); err != nil {
@@ -170,6 +171,31 @@ func handleDaemonModePostCreation(ctx context.Context, cancel context.CancelFunc
 	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
@@ -252,8 +278,6 @@ func startTapperSyncer(ctx context.Context, cancel context.CancelFunc, provider
 		}
 	}()
 
-	state.tapperSyncer = tapperSyncer
-
 	return nil
 }
 
@@ -287,137 +311,6 @@ func readValidationRules(file string) (string, error) {
 	return string(newContent), nil
 }
 
-func createMizuResources(ctx context.Context, cancel context.CancelFunc, kubernetesProvider *kubernetes.Provider, serializedValidationRules string, serializedContract string, serializedMizuConfig string) error {
-	if !config.Config.IsNsRestrictedMode() {
-		if err := createMizuNamespace(ctx, kubernetesProvider); err != nil {
-			return err
-		}
-	}
-
-	if err := createMizuConfigmap(ctx, kubernetesProvider, serializedValidationRules, serializedContract, serializedMizuConfig); err != nil {
-		logger.Log.Warningf(uiUtils.Warning, fmt.Sprintf("Failed to create resources required for policy validation. Mizu will not validate policy rules. error: %v", errormessage.FormatError(err)))
-	}
-
-	var err error
-	state.mizuServiceAccountExists, err = createRBACIfNecessary(ctx, kubernetesProvider)
-	if err != nil {
-		if !config.Config.Tap.DaemonMode {
-			logger.Log.Warningf(uiUtils.Warning, fmt.Sprintf("Failed to ensure the resources required for IP resolving. Mizu will not resolve target IPs to names. error: %v", errormessage.FormatError(err)))
-		}
-	}
-
-	var serviceAccountName string
-	if state.mizuServiceAccountExists {
-		serviceAccountName = kubernetes.ServiceAccountName
-	} else {
-		serviceAccountName = ""
-	}
-
-	opts := &kubernetes.ApiServerOptions{
-		Namespace:             config.Config.MizuResourcesNamespace,
-		PodName:               kubernetes.ApiServerPodName,
-		PodImage:              config.Config.AgentImage,
-		ServiceAccountName:    serviceAccountName,
-		IsNamespaceRestricted: config.Config.IsNsRestrictedMode(),
-		SyncEntriesConfig:     getSyncEntriesConfig(),
-		MaxEntriesDBSizeBytes: config.Config.Tap.MaxEntriesDBSizeBytes(),
-		Resources:             config.Config.Tap.ApiServerResources,
-		ImagePullPolicy:       config.Config.ImagePullPolicy(),
-		LogLevel:              config.Config.LogLevel(),
-	}
-
-	if config.Config.Tap.DaemonMode {
-		if !state.mizuServiceAccountExists {
-			defer cleanUpMizuResources(ctx, cancel, kubernetesProvider)
-			logger.Log.Fatalf(uiUtils.Red, fmt.Sprintf("Failed to ensure the resources required for mizu to run in daemon mode. cannot proceed. error: %v", errormessage.FormatError(err)))
-		}
-		if err := createMizuApiServerDeployment(ctx, kubernetesProvider, opts); err != nil {
-			return err
-		}
-	} else {
-		if err := createMizuApiServerPod(ctx, kubernetesProvider, opts); err != nil {
-			return err
-		}
-	}
-
-	state.apiServerService, err = kubernetesProvider.CreateService(ctx, config.Config.MizuResourcesNamespace, kubernetes.ApiServerPodName, kubernetes.ApiServerPodName)
-	if err != nil {
-		return err
-	}
-	logger.Log.Debugf("Successfully created service: %s", kubernetes.ApiServerPodName)
-
-	return nil
-}
-
-func createMizuConfigmap(ctx context.Context, kubernetesProvider *kubernetes.Provider, serializedValidationRules string, serializedContract string, serializedMizuConfig string) error {
-	err := kubernetesProvider.CreateConfigMap(ctx, config.Config.MizuResourcesNamespace, kubernetes.ConfigMapName, serializedValidationRules, serializedContract, serializedMizuConfig)
-	return err
-}
-
-func createMizuNamespace(ctx context.Context, kubernetesProvider *kubernetes.Provider) error {
-	_, err := kubernetesProvider.CreateNamespace(ctx, config.Config.MizuResourcesNamespace)
-	return err
-}
-
-func createMizuApiServerPod(ctx context.Context, kubernetesProvider *kubernetes.Provider, opts *kubernetes.ApiServerOptions) error {
-	pod, err := kubernetesProvider.GetMizuApiServerPodObject(opts, false, "")
-	if err != nil {
-		return err
-	}
-	if _, err = kubernetesProvider.CreatePod(ctx, config.Config.MizuResourcesNamespace, pod); err != nil {
-		return err
-	}
-	logger.Log.Debugf("Successfully created API server pod: %s", kubernetes.ApiServerPodName)
-	return nil
-}
-
-func createMizuApiServerDeployment(ctx context.Context, kubernetesProvider *kubernetes.Provider, opts *kubernetes.ApiServerOptions) error {
-	volumeClaimCreated := false
-	if !config.Config.Tap.NoPersistentVolumeClaim {
-		volumeClaimCreated = TryToCreatePersistentVolumeClaim(ctx, kubernetesProvider)
-	}
-
-	pod, err := kubernetesProvider.GetMizuApiServerPodObject(opts, volumeClaimCreated, kubernetes.PersistentVolumeClaimName)
-	if err != nil {
-		return err
-	}
-	pod.Spec.Containers[0].LivenessProbe = &core.Probe{
-		Handler: core.Handler{
-			HTTPGet: &core.HTTPGetAction{
-				Path: "/echo",
-				Port: intstr.FromInt(shared.DefaultApiServerPort),
-			},
-		},
-		InitialDelaySeconds: 1,
-		PeriodSeconds:       10,
-	}
-	if _, err = kubernetesProvider.CreateDeployment(ctx, config.Config.MizuResourcesNamespace, opts.PodName, pod); err != nil {
-		return err
-	}
-	logger.Log.Debugf("Successfully created API server deployment: %s", kubernetes.ApiServerPodName)
-	return nil
-}
-
-func TryToCreatePersistentVolumeClaim(ctx context.Context, kubernetesProvider *kubernetes.Provider) bool {
-	isDefaultStorageClassAvailable, err := kubernetesProvider.IsDefaultStorageProviderAvailable(ctx)
-	if err != nil {
-		logger.Log.Warningf(uiUtils.Yellow, "An error occured when checking if a default storage provider exists in this cluster, this means mizu data will be lost on mizu-api-server pod restart")
-		logger.Log.Debugf("error checking if default storage class exists: %v", err)
-		return false
-	} else if !isDefaultStorageClassAvailable {
-		logger.Log.Warningf(uiUtils.Yellow, "Could not find default storage provider in this cluster, this means mizu data will be lost on mizu-api-server pod restart")
-		return false
-	}
-
-	if _, err = kubernetesProvider.CreatePersistentVolumeClaim(ctx, config.Config.MizuResourcesNamespace, kubernetes.PersistentVolumeClaimName, config.Config.Tap.MaxEntriesDBSizeBytes()+mizu.DaemonModePersistentVolumeSizeBufferBytes); err != nil {
-		logger.Log.Warningf(uiUtils.Yellow, "An error has occured while creating a persistent volume claim for mizu, this means mizu data will be lost on mizu-api-server pod restart")
-		logger.Log.Debugf("error creating persistent volume claim: %v", err)
-		return false
-	}
-
-	return true
-}
-
 func getMizuApiFilteringOptions() (*api.TrafficFilteringOptions, error) {
 	var compiledRegexSlice []*api.SerializableRegexp
 
@@ -452,114 +345,6 @@ func getSyncEntriesConfig() *shared.SyncEntriesConfig {
 	}
 }
 
-func cleanUpRestrictedMode(ctx context.Context, kubernetesProvider *kubernetes.Provider) []string {
-	leftoverResources := make([]string, 0)
-
-	if err := kubernetesProvider.RemoveService(ctx, config.Config.MizuResourcesNamespace, kubernetes.ApiServerPodName); err != nil {
-		resourceDesc := fmt.Sprintf("Service %s in namespace %s", kubernetes.ApiServerPodName, config.Config.MizuResourcesNamespace)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	if err := kubernetesProvider.RemoveDaemonSet(ctx, config.Config.MizuResourcesNamespace, kubernetes.TapperDaemonSetName); err != nil {
-		resourceDesc := fmt.Sprintf("DaemonSet %s in namespace %s", kubernetes.TapperDaemonSetName, config.Config.MizuResourcesNamespace)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	if err := kubernetesProvider.RemoveConfigMap(ctx, config.Config.MizuResourcesNamespace, kubernetes.ConfigMapName); err != nil {
-		resourceDesc := fmt.Sprintf("ConfigMap %s in namespace %s", kubernetes.ConfigMapName, config.Config.MizuResourcesNamespace)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	if err := kubernetesProvider.RemoveServicAccount(ctx, config.Config.MizuResourcesNamespace, kubernetes.ServiceAccountName); err != nil {
-		resourceDesc := fmt.Sprintf("Service Account %s in namespace %s", kubernetes.ServiceAccountName, config.Config.MizuResourcesNamespace)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	if err := kubernetesProvider.RemoveRole(ctx, config.Config.MizuResourcesNamespace, kubernetes.RoleName); err != nil {
-		resourceDesc := fmt.Sprintf("Role %s in namespace %s", kubernetes.RoleName, config.Config.MizuResourcesNamespace)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	if err := kubernetesProvider.RemovePod(ctx, config.Config.MizuResourcesNamespace, kubernetes.ApiServerPodName); err != nil {
-		resourceDesc := fmt.Sprintf("Pod %s in namespace %s", kubernetes.ApiServerPodName, config.Config.MizuResourcesNamespace)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	//daemon mode resources
-	if err := kubernetesProvider.RemoveRoleBinding(ctx, config.Config.MizuResourcesNamespace, kubernetes.RoleBindingName); err != nil {
-		resourceDesc := fmt.Sprintf("RoleBinding %s in namespace %s", kubernetes.RoleBindingName, config.Config.MizuResourcesNamespace)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	if err := kubernetesProvider.RemoveDeployment(ctx, config.Config.MizuResourcesNamespace, kubernetes.ApiServerPodName); err != nil {
-		resourceDesc := fmt.Sprintf("Deployment %s in namespace %s", kubernetes.ApiServerPodName, config.Config.MizuResourcesNamespace)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	if err := kubernetesProvider.RemovePersistentVolumeClaim(ctx, config.Config.MizuResourcesNamespace, kubernetes.PersistentVolumeClaimName); err != nil {
-		resourceDesc := fmt.Sprintf("PersistentVolumeClaim %s in namespace %s", kubernetes.PersistentVolumeClaimName, config.Config.MizuResourcesNamespace)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	if err := kubernetesProvider.RemoveRole(ctx, config.Config.MizuResourcesNamespace, kubernetes.DaemonRoleName); err != nil {
-		resourceDesc := fmt.Sprintf("Role %s in namespace %s", kubernetes.DaemonRoleName, config.Config.MizuResourcesNamespace)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	if err := kubernetesProvider.RemoveRoleBinding(ctx, config.Config.MizuResourcesNamespace, kubernetes.DaemonRoleBindingName); err != nil {
-		resourceDesc := fmt.Sprintf("RoleBinding %s in namespace %s", kubernetes.DaemonRoleBindingName, config.Config.MizuResourcesNamespace)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	return leftoverResources
-}
-
-func cleanUpNonRestrictedMode(ctx context.Context, cancel context.CancelFunc, kubernetesProvider *kubernetes.Provider) []string {
-	leftoverResources := make([]string, 0)
-
-	if err := kubernetesProvider.RemoveNamespace(ctx, config.Config.MizuResourcesNamespace); err != nil {
-		resourceDesc := fmt.Sprintf("Namespace %s", config.Config.MizuResourcesNamespace)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	} else {
-		defer waitUntilNamespaceDeleted(ctx, cancel, kubernetesProvider)
-	}
-
-	if err := kubernetesProvider.RemoveClusterRole(ctx, kubernetes.ClusterRoleName); err != nil {
-		resourceDesc := fmt.Sprintf("ClusterRole %s", kubernetes.ClusterRoleName)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	if err := kubernetesProvider.RemoveClusterRoleBinding(ctx, kubernetes.ClusterRoleBindingName); err != nil {
-		resourceDesc := fmt.Sprintf("ClusterRoleBinding %s", kubernetes.ClusterRoleBindingName)
-		handleDeletionError(err, resourceDesc, &leftoverResources)
-	}
-
-	return leftoverResources
-}
-
-func handleDeletionError(err error, resourceDesc string, leftoverResources *[]string) {
-	logger.Log.Debugf("Error removing %s: %v", resourceDesc, errormessage.FormatError(err))
-	*leftoverResources = append(*leftoverResources, resourceDesc)
-}
-
-func waitUntilNamespaceDeleted(ctx context.Context, cancel context.CancelFunc, kubernetesProvider *kubernetes.Provider) {
-	// Call cancel if a terminating signal was received. Allows user to skip the wait.
-	go func() {
-		waitForFinish(ctx, cancel)
-	}()
-
-	if err := kubernetesProvider.WaitUtilNamespaceDeleted(ctx, config.Config.MizuResourcesNamespace); err != nil {
-		switch {
-		case ctx.Err() == context.Canceled:
-			logger.Log.Debugf("Do nothing. User interrupted the wait")
-		case err == wait.ErrWaitTimeout:
-			logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Timeout while removing Namespace %s", config.Config.MizuResourcesNamespace))
-		default:
-			logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Error while waiting for Namespace %s to be deleted: %v", config.Config.MizuResourcesNamespace, errormessage.FormatError(err)))
-		}
-	}
-}
-
 func watchApiServerPod(ctx context.Context, kubernetesProvider *kubernetes.Provider, cancel context.CancelFunc) {
 	podExactRegex := regexp.MustCompile(fmt.Sprintf("^%s$", kubernetes.ApiServerPodName))
 	podWatchHelper := kubernetes.NewPodWatchHelper(kubernetesProvider, podExactRegex)
@@ -635,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) {
@@ -663,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
@@ -705,21 +491,3 @@ func getNamespaces(kubernetesProvider *kubernetes.Provider) []string {
 		return []string{currentNamespace}
 	}
 }
-
-func createRBACIfNecessary(ctx context.Context, kubernetesProvider *kubernetes.Provider) (bool, error) {
-	if !config.Config.IsNsRestrictedMode() {
-		if err := kubernetesProvider.CreateMizuRBAC(ctx, config.Config.MizuResourcesNamespace, kubernetes.ServiceAccountName, kubernetes.ClusterRoleName, kubernetes.ClusterRoleBindingName, mizu.RBACVersion); err != nil {
-			return false, err
-		}
-	} else {
-		if err := kubernetesProvider.CreateMizuRBACNamespaceRestricted(ctx, config.Config.MizuResourcesNamespace, kubernetes.ServiceAccountName, kubernetes.RoleName, kubernetes.RoleBindingName, mizu.RBACVersion); err != nil {
-			return false, err
-		}
-	}
-	if config.Config.Tap.DaemonMode {
-		if err := kubernetesProvider.CreateDaemonsetRBAC(ctx, config.Config.MizuResourcesNamespace, kubernetes.ServiceAccountName, kubernetes.DaemonRoleName, kubernetes.DaemonRoleBindingName, mizu.RBACVersion); err != nil {
-			return false, err
-		}
-	}
-	return true, nil
-}

cli/cmd/viewRunner.go

@@ -3,6 +3,7 @@ package cmd
 import (
 	"context"
 	"fmt"
+	"github.com/up9inc/mizu/cli/utils"
 	"net/http"
 
 	"github.com/up9inc/mizu/cli/apiserver"
@@ -71,5 +72,5 @@ func runMizuView() {
 		return
 	}
 
-	waitForFinish(ctx, cancel)
+	utils.WaitForFinish(ctx, cancel)
 }

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(),
-		DaemonMode:              Config.Tap.DaemonMode,
-		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,
-	}
-	return &config, nil
-}

cli/resources/cleanResources.go

@@ -0,0 +1,141 @@
+package resources
+
+import (
+	"context"
+	"fmt"
+	"github.com/up9inc/mizu/cli/errormessage"
+	"github.com/up9inc/mizu/cli/mizu/fsUtils"
+	"github.com/up9inc/mizu/cli/uiUtils"
+	"github.com/up9inc/mizu/cli/utils"
+	"github.com/up9inc/mizu/shared/kubernetes"
+	"github.com/up9inc/mizu/shared/logger"
+	"k8s.io/apimachinery/pkg/util/wait"
+)
+
+func CleanUpMizuResources(ctx context.Context, cancel context.CancelFunc, kubernetesProvider *kubernetes.Provider, isNsRestrictedMode bool, mizuResourcesNamespace string) {
+	logger.Log.Infof("\nRemoving mizu resources")
+
+	var leftoverResources []string
+
+	if isNsRestrictedMode {
+		leftoverResources = cleanUpRestrictedMode(ctx, kubernetesProvider, mizuResourcesNamespace)
+	} else {
+		leftoverResources = cleanUpNonRestrictedMode(ctx, cancel, kubernetesProvider, mizuResourcesNamespace)
+	}
+
+	if len(leftoverResources) > 0 {
+		errMsg := fmt.Sprintf("Failed to remove the following resources, for more info check logs at %s:", fsUtils.GetLogFilePath())
+		for _, resource := range leftoverResources {
+			errMsg += "\n- " + resource
+		}
+		logger.Log.Errorf(uiUtils.Error, errMsg)
+	}
+}
+
+func cleanUpNonRestrictedMode(ctx context.Context, cancel context.CancelFunc, kubernetesProvider *kubernetes.Provider, mizuResourcesNamespace string) []string {
+	leftoverResources := make([]string, 0)
+
+	if err := kubernetesProvider.RemoveNamespace(ctx, mizuResourcesNamespace); err != nil {
+		resourceDesc := fmt.Sprintf("Namespace %s", mizuResourcesNamespace)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	} else {
+		defer waitUntilNamespaceDeleted(ctx, cancel, kubernetesProvider, mizuResourcesNamespace)
+	}
+
+	if err := kubernetesProvider.RemoveClusterRole(ctx, kubernetes.ClusterRoleName); err != nil {
+		resourceDesc := fmt.Sprintf("ClusterRole %s", kubernetes.ClusterRoleName)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	if err := kubernetesProvider.RemoveClusterRoleBinding(ctx, kubernetes.ClusterRoleBindingName); err != nil {
+		resourceDesc := fmt.Sprintf("ClusterRoleBinding %s", kubernetes.ClusterRoleBindingName)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	return leftoverResources
+}
+
+func waitUntilNamespaceDeleted(ctx context.Context, cancel context.CancelFunc, kubernetesProvider *kubernetes.Provider, mizuResourcesNamespace string) {
+	// Call cancel if a terminating signal was received. Allows user to skip the wait.
+	go func() {
+		utils.WaitForFinish(ctx, cancel)
+	}()
+
+	if err := kubernetesProvider.WaitUtilNamespaceDeleted(ctx, mizuResourcesNamespace); err != nil {
+		switch {
+		case ctx.Err() == context.Canceled:
+			logger.Log.Debugf("Do nothing. User interrupted the wait")
+		case err == wait.ErrWaitTimeout:
+			logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Timeout while removing Namespace %s", mizuResourcesNamespace))
+		default:
+			logger.Log.Errorf(uiUtils.Error, fmt.Sprintf("Error while waiting for Namespace %s to be deleted: %v", mizuResourcesNamespace, errormessage.FormatError(err)))
+		}
+	}
+}
+
+func cleanUpRestrictedMode(ctx context.Context, kubernetesProvider *kubernetes.Provider, mizuResourcesNamespace string) []string {
+	leftoverResources := make([]string, 0)
+
+	if err := kubernetesProvider.RemoveService(ctx, mizuResourcesNamespace, kubernetes.ApiServerPodName); err != nil {
+		resourceDesc := fmt.Sprintf("Service %s in namespace %s", kubernetes.ApiServerPodName, mizuResourcesNamespace)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	if err := kubernetesProvider.RemoveDaemonSet(ctx, mizuResourcesNamespace, kubernetes.TapperDaemonSetName); err != nil {
+		resourceDesc := fmt.Sprintf("DaemonSet %s in namespace %s", kubernetes.TapperDaemonSetName, mizuResourcesNamespace)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	if err := kubernetesProvider.RemoveConfigMap(ctx, mizuResourcesNamespace, kubernetes.ConfigMapName); err != nil {
+		resourceDesc := fmt.Sprintf("ConfigMap %s in namespace %s", kubernetes.ConfigMapName, mizuResourcesNamespace)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	if err := kubernetesProvider.RemoveServicAccount(ctx, mizuResourcesNamespace, kubernetes.ServiceAccountName); err != nil {
+		resourceDesc := fmt.Sprintf("Service Account %s in namespace %s", kubernetes.ServiceAccountName, mizuResourcesNamespace)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	if err := kubernetesProvider.RemoveRole(ctx, mizuResourcesNamespace, kubernetes.RoleName); err != nil {
+		resourceDesc := fmt.Sprintf("Role %s in namespace %s", kubernetes.RoleName, mizuResourcesNamespace)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	if err := kubernetesProvider.RemovePod(ctx, mizuResourcesNamespace, kubernetes.ApiServerPodName); err != nil {
+		resourceDesc := fmt.Sprintf("Pod %s in namespace %s", kubernetes.ApiServerPodName, mizuResourcesNamespace)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	//install mode resources
+	if err := kubernetesProvider.RemoveRoleBinding(ctx, mizuResourcesNamespace, kubernetes.RoleBindingName); err != nil {
+		resourceDesc := fmt.Sprintf("RoleBinding %s in namespace %s", kubernetes.RoleBindingName, mizuResourcesNamespace)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	if err := kubernetesProvider.RemoveDeployment(ctx, mizuResourcesNamespace, kubernetes.ApiServerPodName); err != nil {
+		resourceDesc := fmt.Sprintf("Deployment %s in namespace %s", kubernetes.ApiServerPodName, mizuResourcesNamespace)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	if err := kubernetesProvider.RemovePersistentVolumeClaim(ctx, mizuResourcesNamespace, kubernetes.PersistentVolumeClaimName); err != nil {
+		resourceDesc := fmt.Sprintf("PersistentVolumeClaim %s in namespace %s", kubernetes.PersistentVolumeClaimName, mizuResourcesNamespace)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	if err := kubernetesProvider.RemoveRole(ctx, mizuResourcesNamespace, kubernetes.DaemonRoleName); err != nil {
+		resourceDesc := fmt.Sprintf("Role %s in namespace %s", kubernetes.DaemonRoleName, mizuResourcesNamespace)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	if err := kubernetesProvider.RemoveRoleBinding(ctx, mizuResourcesNamespace, kubernetes.DaemonRoleBindingName); err != nil {
+		resourceDesc := fmt.Sprintf("RoleBinding %s in namespace %s", kubernetes.DaemonRoleBindingName, mizuResourcesNamespace)
+		handleDeletionError(err, resourceDesc, &leftoverResources)
+	}
+
+	return leftoverResources
+}
+
+func handleDeletionError(err error, resourceDesc string, leftoverResources *[]string) {
+	logger.Log.Debugf("Error removing %s: %v", resourceDesc, errormessage.FormatError(err))
+	*leftoverResources = append(*leftoverResources, resourceDesc)
+}

cli/resources/createResources.go

@@ -0,0 +1,163 @@
+package resources
+
+import (
+	"context"
+	"fmt"
+	"github.com/op/go-logging"
+	"github.com/up9inc/mizu/cli/errormessage"
+	"github.com/up9inc/mizu/cli/mizu"
+	"github.com/up9inc/mizu/cli/uiUtils"
+	"github.com/up9inc/mizu/shared"
+	"github.com/up9inc/mizu/shared/kubernetes"
+	"github.com/up9inc/mizu/shared/logger"
+	core "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
+)
+
+func CreateMizuResources(ctx context.Context, cancel context.CancelFunc, kubernetesProvider *kubernetes.Provider, serializedValidationRules string, serializedContract string, serializedMizuConfig string, isNsRestrictedMode bool, mizuResourcesNamespace string, isInstallMode bool, agentImage string, syncEntriesConfig *shared.SyncEntriesConfig, maxEntriesDBSizeBytes int64, apiServerResources shared.Resources, imagePullPolicy core.PullPolicy, logLevel logging.Level, noPersistentVolumeClaim bool) (bool, error) {
+	if !isNsRestrictedMode {
+		if err := createMizuNamespace(ctx, kubernetesProvider, mizuResourcesNamespace); err != nil {
+			return false, err
+		}
+	}
+
+	if err := createMizuConfigmap(ctx, kubernetesProvider, serializedValidationRules, serializedContract, serializedMizuConfig, mizuResourcesNamespace); err != nil {
+		logger.Log.Warningf(uiUtils.Warning, fmt.Sprintf("Failed to create resources required for policy validation. Mizu will not validate policy rules. error: %v", errormessage.FormatError(err)))
+	}
+
+	mizuServiceAccountExists, err := createRBACIfNecessary(ctx, kubernetesProvider, isNsRestrictedMode, mizuResourcesNamespace, isInstallMode)
+	if err != nil {
+		if !isInstallMode {
+			logger.Log.Warningf(uiUtils.Warning, fmt.Sprintf("Failed to ensure the resources required for IP resolving. Mizu will not resolve target IPs to names. error: %v", errormessage.FormatError(err)))
+		}
+	}
+
+	var serviceAccountName string
+	if mizuServiceAccountExists {
+		serviceAccountName = kubernetes.ServiceAccountName
+	} else {
+		serviceAccountName = ""
+	}
+
+	opts := &kubernetes.ApiServerOptions{
+		Namespace:             mizuResourcesNamespace,
+		PodName:               kubernetes.ApiServerPodName,
+		PodImage:              agentImage,
+		ServiceAccountName:    serviceAccountName,
+		IsNamespaceRestricted: isNsRestrictedMode,
+		SyncEntriesConfig:     syncEntriesConfig,
+		MaxEntriesDBSizeBytes: maxEntriesDBSizeBytes,
+		Resources:             apiServerResources,
+		ImagePullPolicy:       imagePullPolicy,
+		LogLevel:              logLevel,
+	}
+
+	if isInstallMode {
+		if !mizuServiceAccountExists {
+			defer CleanUpMizuResources(ctx, cancel, kubernetesProvider, isNsRestrictedMode, mizuResourcesNamespace)
+			logger.Log.Fatalf(uiUtils.Red, fmt.Sprintf("Failed to ensure the resources required for mizu to run in daemon mode. cannot proceed. error: %v", errormessage.FormatError(err)))
+		}
+		if err := createMizuApiServerDeployment(ctx, kubernetesProvider, opts, noPersistentVolumeClaim); err != nil {
+			return mizuServiceAccountExists, err
+		}
+	} else {
+		if err := createMizuApiServerPod(ctx, kubernetesProvider, opts); err != nil {
+			return mizuServiceAccountExists, err
+		}
+	}
+
+	_, err = kubernetesProvider.CreateService(ctx, mizuResourcesNamespace, kubernetes.ApiServerPodName, kubernetes.ApiServerPodName)
+	if err != nil {
+		return mizuServiceAccountExists, err
+	}
+	logger.Log.Debugf("Successfully created service: %s", kubernetes.ApiServerPodName)
+
+	return mizuServiceAccountExists, nil
+}
+
+func createMizuNamespace(ctx context.Context, kubernetesProvider *kubernetes.Provider, mizuResourcesNamespace string) error {
+	_, err := kubernetesProvider.CreateNamespace(ctx, mizuResourcesNamespace)
+	return err
+}
+
+func createMizuConfigmap(ctx context.Context, kubernetesProvider *kubernetes.Provider, serializedValidationRules string, serializedContract string, serializedMizuConfig string, mizuResourcesNamespace string) error {
+	err := kubernetesProvider.CreateConfigMap(ctx, mizuResourcesNamespace, kubernetes.ConfigMapName, serializedValidationRules, serializedContract, serializedMizuConfig)
+	return err
+}
+
+func createRBACIfNecessary(ctx context.Context, kubernetesProvider *kubernetes.Provider, isNsRestrictedMode bool, mizuResourcesNamespace string, isInstallMode bool) (bool, error) {
+	if !isNsRestrictedMode {
+		if err := kubernetesProvider.CreateMizuRBAC(ctx, mizuResourcesNamespace, kubernetes.ServiceAccountName, kubernetes.ClusterRoleName, kubernetes.ClusterRoleBindingName, mizu.RBACVersion); err != nil {
+			return false, err
+		}
+	} else {
+		if err := kubernetesProvider.CreateMizuRBACNamespaceRestricted(ctx, mizuResourcesNamespace, kubernetes.ServiceAccountName, kubernetes.RoleName, kubernetes.RoleBindingName, mizu.RBACVersion); err != nil {
+			return false, err
+		}
+	}
+	if isInstallMode {
+		if err := kubernetesProvider.CreateDaemonsetRBAC(ctx, mizuResourcesNamespace, kubernetes.ServiceAccountName, kubernetes.DaemonRoleName, kubernetes.DaemonRoleBindingName, mizu.RBACVersion); err != nil {
+			return false, err
+		}
+	}
+	return true, nil
+}
+
+func createMizuApiServerDeployment(ctx context.Context, kubernetesProvider *kubernetes.Provider, opts *kubernetes.ApiServerOptions, noPersistentVolumeClaim bool) error {
+	volumeClaimCreated := false
+	if !noPersistentVolumeClaim {
+		volumeClaimCreated = tryToCreatePersistentVolumeClaim(ctx, kubernetesProvider, opts)
+	}
+
+	pod, err := kubernetesProvider.GetMizuApiServerPodObject(opts, volumeClaimCreated, kubernetes.PersistentVolumeClaimName)
+	if err != nil {
+		return err
+	}
+	pod.Spec.Containers[0].LivenessProbe = &core.Probe{
+		Handler: core.Handler{
+			HTTPGet: &core.HTTPGetAction{
+				Path: "/echo",
+				Port: intstr.FromInt(shared.DefaultApiServerPort),
+			},
+		},
+		InitialDelaySeconds: 1,
+		PeriodSeconds:       10,
+	}
+	if _, err = kubernetesProvider.CreateDeployment(ctx, opts.Namespace, opts.PodName, pod); err != nil {
+		return err
+	}
+	logger.Log.Debugf("Successfully created API server deployment: %s", kubernetes.ApiServerPodName)
+	return nil
+}
+
+func tryToCreatePersistentVolumeClaim(ctx context.Context, kubernetesProvider *kubernetes.Provider, opts *kubernetes.ApiServerOptions) bool {
+	isDefaultStorageClassAvailable, err := kubernetesProvider.IsDefaultStorageProviderAvailable(ctx)
+	if err != nil {
+		logger.Log.Warningf(uiUtils.Yellow, "An error occured when checking if a default storage provider exists in this cluster, this means mizu data will be lost on mizu-api-server pod restart")
+		logger.Log.Debugf("error checking if default storage class exists: %v", err)
+		return false
+	} else if !isDefaultStorageClassAvailable {
+		logger.Log.Warningf(uiUtils.Yellow, "Could not find default storage provider in this cluster, this means mizu data will be lost on mizu-api-server pod restart")
+		return false
+	}
+
+	if _, err = kubernetesProvider.CreatePersistentVolumeClaim(ctx, opts.Namespace, kubernetes.PersistentVolumeClaimName, opts.MaxEntriesDBSizeBytes + mizu.DaemonModePersistentVolumeSizeBufferBytes); err != nil {
+		logger.Log.Warningf(uiUtils.Yellow, "An error has occured while creating a persistent volume claim for mizu, this means mizu data will be lost on mizu-api-server pod restart")
+		logger.Log.Debugf("error creating persistent volume claim: %v", err)
+		return false
+	}
+
+	return true
+}
+
+func createMizuApiServerPod(ctx context.Context, kubernetesProvider *kubernetes.Provider, opts *kubernetes.ApiServerOptions) error {
+	pod, err := kubernetesProvider.GetMizuApiServerPodObject(opts, false, "")
+	if err != nil {
+		return err
+	}
+	if _, err = kubernetesProvider.CreatePod(ctx, opts.Namespace, pod); err != nil {
+		return err
+	}
+	logger.Log.Debugf("Successfully created API server pod: %s", kubernetes.ApiServerPodName)
+	return nil
+}

cli/utils/waitUtils.go

@@ -0,0 +1,25 @@
+package utils
+
+import (
+	"context"
+	"github.com/up9inc/mizu/shared/logger"
+	"os"
+	"os/signal"
+	"syscall"
+)
+
+func WaitForFinish(ctx context.Context, cancel context.CancelFunc) {
+	logger.Log.Debugf("waiting for finish...")
+	sigChan := make(chan os.Signal, 1)
+	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
+
+	// block until ctx cancel is called or termination signal is received
+	select {
+	case <-ctx.Done():
+		logger.Log.Debugf("ctx done")
+		break
+	case <-sigChan:
+		logger.Log.Debugf("Got termination signal, canceling execution...")
+		cancel()
+	}
+}

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")

shared/models.go

@@ -1,11 +1,13 @@
 package shared
 
 import (
+	"io/ioutil"
+	"strings"
+
 	"github.com/op/go-logging"
 	"github.com/up9inc/mizu/shared/logger"
 	"github.com/up9inc/mizu/tap/api"
-	"io/ioutil"
-	"strings"
+	v1 "k8s.io/api/core/v1"
 
 	"gopkg.in/yaml.v3"
 )
@@ -21,6 +23,7 @@ const (
 	WebSocketMessageTypeToast         WebSocketMessageType = "toast"
 	WebSocketMessageTypeQueryMetadata WebSocketMessageType = "queryMetadata"
 	WebSocketMessageTypeStartTime     WebSocketMessageType = "startTime"
+	WebSocketMessageTypeTapConfig     WebSocketMessageType = "tapConfig"
 )
 
 type Resources struct {
@@ -33,7 +36,6 @@ type Resources struct {
 type MizuAgentConfig struct {
 	TapTargetRegex          api.SerializableRegexp      `json:"tapTargetRegex"`
 	MaxDBSizeBytes          int64                       `json:"maxDBSizeBytes"`
-	DaemonMode              bool                        `json:"daemonMode"`
 	TargetNamespaces        []string                    `json:"targetNamespaces"`
 	AgentImage              string                      `json:"agentImage"`
 	PullPolicy              string                      `json:"pullPolicy"`
@@ -44,6 +46,7 @@ type MizuAgentConfig struct {
 	MizuApiFilteringOptions api.TrafficFilteringOptions `json:"mizuApiFilteringOptions"`
 	AgentDatabasePath       string                      `json:"agentDatabasePath"`
 	Istio                   bool                        `json:"istio"`
+	SyncTappers             bool                        `json:"syncTappers"`
 }
 
 type WebSocketMessageMetadata struct {
@@ -67,6 +70,11 @@ type WebSocketStatusMessage struct {
 	TappingStatus []TappedPodStatus `json:"tappingStatus"`
 }
 
+type WebSocketTapConfigMessage struct {
+	*WebSocketMessageMetadata
+	TapTargets []v1.Pod `json:"pods"`
+}
+
 type TapperStatus struct {
 	TapperName string `json:"tapperName"`
 	NodeName   string `json:"nodeName"`

tap/passive_tapper.go

@@ -11,6 +11,7 @@ package tap
 import (
 	"encoding/json"
 	"flag"
+	"fmt"
 	"os"
 	"runtime"
 	"strings"
@@ -60,8 +61,11 @@ type TapOpts struct {
 	FilterAuthorities []v1.Pod
 }
 
-var extensions []*api.Extension                   // global
-var filteringOptions *api.TrafficFilteringOptions // global
+var extensions []*api.Extension                     // global
+var filteringOptions *api.TrafficFilteringOptions   // global
+var tapTargets []v1.Pod                             // global
+var packetSourceManager *source.PacketSourceManager // global
+var mainPacketInputChan chan source.TcpPacketInfo   // global
 
 func inArrayInt(arr []int, valueToCheck int) bool {
 	for _, value := range arr {
@@ -86,7 +90,9 @@ func StartPassiveTapper(opts *TapOpts, outputItems chan *api.OutputChannelItem,
 	filteringOptions = options
 
 	if opts.FilterAuthorities == nil {
-		opts.FilterAuthorities = []v1.Pod{}
+		tapTargets = []v1.Pod{}
+	} else {
+		tapTargets = opts.FilterAuthorities
 	}
 
 	if GetMemoryProfilingEnabled() {
@@ -96,6 +102,23 @@ func StartPassiveTapper(opts *TapOpts, outputItems chan *api.OutputChannelItem,
 	go startPassiveTapper(opts, outputItems)
 }
 
+func UpdateTapTargets(newTapTargets []v1.Pod) {
+	tapTargets = newTapTargets
+	if err := initializePacketSources(); err != nil {
+		logger.Log.Fatal(err)
+	}
+	printNewTapTargets()
+}
+
+func printNewTapTargets() {
+	printStr := ""
+	for _, tapTarget := range tapTargets {
+		printStr += fmt.Sprintf("%s (%s), ", tapTarget.Status.PodIP, tapTarget.Name)
+	}
+	printStr = strings.TrimRight(printStr, ", ")
+	logger.Log.Infof("Now tapping: %s", printStr)
+}
+
 func printPeriodicStats(cleaner *Cleaner) {
 	statsPeriod := time.Second * time.Duration(*statsevery)
 	ticker := time.NewTicker(statsPeriod)
@@ -136,7 +159,11 @@ func printPeriodicStats(cleaner *Cleaner) {
 	}
 }
 
-func initializePacketSources(opts *TapOpts) (*source.PacketSourceManager, error) {
+func initializePacketSources() error {
+	if packetSourceManager != nil {
+		packetSourceManager.Close()
+	}
+
 	var bpffilter string
 	if len(flag.Args()) > 0 {
 		bpffilter = strings.Join(flag.Args(), " ")
@@ -151,7 +178,13 @@ func initializePacketSources(opts *TapOpts) (*source.PacketSourceManager, error)
 		BpfFilter:   bpffilter,
 	}
 
-	return source.NewPacketSourceManager(*procfs, *pids, *fname, *iface, *istio, opts.FilterAuthorities, behaviour)
+	var err error
+	if packetSourceManager, err = source.NewPacketSourceManager(*procfs, *pids, *fname, *iface, *istio, tapTargets, behaviour); err != nil {
+		return err
+	} else {
+		packetSourceManager.ReadPackets(!*nodefrag, mainPacketInputChan)
+		return nil
+	}
 }
 
 func startPassiveTapper(opts *TapOpts, outputItems chan *api.OutputChannelItem) {
@@ -161,25 +194,16 @@ func startPassiveTapper(opts *TapOpts, outputItems chan *api.OutputChannelItem)
 	diagnose.InitializeErrorsMap(*debug, *verbose, *quiet)
 	diagnose.InitializeTapperInternalStats()
 
-	sources, err := initializePacketSources(opts)
+	mainPacketInputChan = make(chan source.TcpPacketInfo)
 
-	if err != nil {
+	if err := initializePacketSources(); err != nil {
 		logger.Log.Fatal(err)
 	}
 
-	defer sources.Close()
-
-	if err != nil {
-		logger.Log.Fatal(err)
-	}
-
-	packets := make(chan source.TcpPacketInfo)
 	assembler := NewTcpAssembler(outputItems, streamsMap, opts)
 
 	diagnose.AppStats.SetStartTime(time.Now())
 
-	sources.ReadPackets(!*nodefrag, packets)
-
 	staleConnectionTimeout := time.Second * time.Duration(*staleTimeoutSeconds)
 	cleaner := Cleaner{
 		assembler:         assembler.Assembler,
@@ -191,7 +215,7 @@ func startPassiveTapper(opts *TapOpts, outputItems chan *api.OutputChannelItem)
 
 	go printPeriodicStats(&cleaner)
 
-	assembler.processPackets(*hexdumppkt, packets)
+	assembler.processPackets(*hexdumppkt, mainPacketInputChan)
 
 	if diagnose.TapErrors.OutputLevel >= 2 {
 		assembler.dumpStreamPool()

tap/tcp_stream_factory.go

@@ -151,13 +151,13 @@ func inArrayPod(pods []v1.Pod, address string) bool {
 
 func (factory *tcpStreamFactory) getStreamProps(srcIP string, srcPort string, dstIP string, dstPort string) *streamProps {
 	if factory.opts.HostMode {
-		if inArrayPod(factory.opts.FilterAuthorities, fmt.Sprintf("%s:%s", dstIP, dstPort)) {
+		if inArrayPod(tapTargets, fmt.Sprintf("%s:%s", dstIP, dstPort)) {
 			return &streamProps{isTapTarget: true, isOutgoing: false}
-		} else if inArrayPod(factory.opts.FilterAuthorities, dstIP) {
+		} else if inArrayPod(tapTargets, dstIP) {
 			return &streamProps{isTapTarget: true, isOutgoing: false}
-		} else if inArrayPod(factory.opts.FilterAuthorities, fmt.Sprintf("%s:%s", srcIP, srcPort)) {
+		} else if inArrayPod(tapTargets, fmt.Sprintf("%s:%s", srcIP, srcPort)) {
 			return &streamProps{isTapTarget: true, isOutgoing: true}
-		} else if inArrayPod(factory.opts.FilterAuthorities, srcIP) {
+		} else if inArrayPod(tapTargets, srcIP) {
 			return &streamProps{isTapTarget: true, isOutgoing: true}
 		}
 		return &streamProps{isTapTarget: false, isOutgoing: false}