Support stopping oas/servicemesh/telemetry in flight (#869)

* Update oas_generator.go and servicemap.go

* Update oas_generator.go

* Update esClient.go

* Update servicemap.go

.github/workflows/acceptance_tests.yml

@@ -24,6 +24,16 @@ jobs:
       - name: Setup acceptance test
         run: source ./acceptanceTests/setup.sh
 
+      - name: Create k8s users and change context
+        env:
+          USERNAME_UNRESTRICTED: user-with-clusterwide-access
+          USERNAME_RESTRICTED: user-with-restricted-access
+        run: |
+          ./acceptanceTests/create_user.sh "${USERNAME_UNRESTRICTED}"
+          ./acceptanceTests/create_user.sh "${USERNAME_RESTRICTED}"
+          kubectl apply -f cli/cmd/permissionFiles/permissions-all-namespaces-tap.yaml
+          kubectl config use-context ${USERNAME_UNRESTRICTED}
+
       - name: Test
         run: make acceptance-test
 

acceptanceTests/create_user.sh

@@ -0,0 +1,37 @@
+#!/bin/bash
+
+# Create a user in Minikube cluster "minikube"
+# Create context for user
+# Usage:
+#  ./create_user.sh <username>
+
+set -e
+
+NEW_USERNAME=$1
+CERT_DIR="${HOME}/certs"
+KEY_FILE="${CERT_DIR}/${NEW_USERNAME}.key"
+CRT_FILE="${CERT_DIR}/${NEW_USERNAME}.crt"
+MINIKUBE_KEY_FILE="${HOME}/.minikube/ca.key"
+MINIKUBE_CRT_FILE="${HOME}/.minikube/ca.crt"
+DAYS=1
+
+echo "Creating user and context for username \"${NEW_USERNAME}\" in Minikube cluster"
+
+if ! command -v openssl &> /dev/null
+then
+	 echo "Installing openssl"
+	 sudo apt-get update
+	 sudo apt-get install openssl
+fi
+
+echo "Creating certificate for user \"${NEW_USERNAME}\""
+mkdir -p ${CERT_DIR}
+echo "Generating key \"${KEY_FILE}\""
+openssl genrsa -out "${KEY_FILE}" 2048
+echo "Generating crt \"${CRT_FILE}\""
+openssl req -new -key "${KEY_FILE}" -out "${CRT_FILE}" -subj "/CN=${NEW_USERNAME}/O=group1"
+openssl x509 -req -in "${CRT_FILE}" -CA "${MINIKUBE_CRT_FILE}" -CAkey "${MINIKUBE_KEY_FILE}" -CAcreateserial -out "${CRT_FILE}" -days $DAYS
+
+echo "Creating context for user \"${NEW_USERNAME}\""
+kubectl config set-credentials "${NEW_USERNAME}" --client-certificate="${CRT_FILE}" --client-key="${KEY_FILE}"
+kubectl config set-context "${NEW_USERNAME}" --cluster=minikube --user="${NEW_USERNAME}"

acceptanceTests/extensions_test.go

@@ -49,7 +49,13 @@ func TestRedis(t *testing.T) {
 
 	ctx := context.Background()
 
-	redisExternalIp, err := GetServiceExternalIp(ctx, defaultNamespaceName, "redis")
+	kubernetesProvider, err := NewKubernetesProvider()
+	if err != nil {
+		t.Errorf("failed to create k8s provider, err %v", err)
+		return
+	}
+
+	redisExternalIp, err := kubernetesProvider.GetServiceExternalIp(ctx, defaultNamespaceName, "redis")
 	if err != nil {
 		t.Errorf("failed to get redis external ip, err: %v", err)
 		return
@@ -141,7 +147,13 @@ func TestAmqp(t *testing.T) {
 
 	ctx := context.Background()
 
-	rabbitmqExternalIp, err := GetServiceExternalIp(ctx, defaultNamespaceName, "rabbitmq")
+	kubernetesProvider, err := NewKubernetesProvider()
+	if err != nil {
+		t.Errorf("failed to create k8s provider, err %v", err)
+		return
+	}
+
+	rabbitmqExternalIp, err := kubernetesProvider.GetServiceExternalIp(ctx, defaultNamespaceName, "rabbitmq")
 	if err != nil {
 		t.Errorf("failed to get RabbitMQ external ip, err: %v", err)
 		return

acceptanceTests/setup.sh

@@ -57,9 +57,6 @@ kubectl expose deployment rabbitmq --type=LoadBalancer --port=5672 -n mizu-tests
 echo "Starting proxy"
 kubectl proxy --port=8080 &
 
-echo "Starting tunnel"
-minikube tunnel &
-
 echo "Setting minikube docker env"
 eval $(minikube docker-env)
 
@@ -68,3 +65,6 @@ make build-docker-ci
 
 echo "Build cli"
 make build-cli-ci
+
+echo "Starting tunnel"
+minikube tunnel &

acceptanceTests/tap_test.go

@@ -14,6 +14,10 @@ import (
 )
 
 func TestTap(t *testing.T) {
+	basicTapTest(t, false)
+}
+
+func basicTapTest(t *testing.T, shouldCheckSrcAndDest bool, extraArgs... string) {
 	if testing.Short() {
 		t.Skip("ignored acceptance test")
 	}
@@ -33,6 +37,8 @@ func TestTap(t *testing.T) {
 			tapNamespace := GetDefaultTapNamespace()
 			tapCmdArgs = append(tapCmdArgs, tapNamespace...)
 
+			tapCmdArgs = append(tapCmdArgs, extraArgs...)
+
 			tapCmd := exec.Command(cliPath, tapCmdArgs...)
 			t.Logf("running command: %v", tapCmd.String())
 
@@ -72,7 +78,6 @@ func TestTap(t *testing.T) {
 				expectedPodsStr += fmt.Sprintf("Name:%vNamespace:%v", expectedPods[i].Name, expectedPods[i].Namespace)
 			}
 
-			const shouldCheckSrcAndDest = false
 			RunCypressTests(t, fmt.Sprintf("npx cypress run --spec  \"cypress/integration/tests/UiTest.js\" --env entriesCount=%d,arrayDict=%v,shouldCheckSrcAndDest=%v",
 				entriesCount, expectedPodsStr, shouldCheckSrcAndDest))
 		})
@@ -644,3 +649,44 @@ func TestTapDumpLogs(t *testing.T) {
 		return
 	}
 }
+
+func TestIpResolving(t *testing.T) {
+	namespace := allNamespaces
+
+	t.Log("add permissions for ip-resolution for current user")
+	if err := ApplyKubeFilesForTest(
+		t,
+		"minikube",
+		namespace,
+		"../cli/cmd/permissionFiles/permissions-all-namespaces-ip-resolution-optional.yaml",
+	); err != nil {
+		t.Errorf("failed to create k8s permissions, %v", err)
+		return
+	}
+
+	basicTapTest(t, true)
+}
+
+func TestRestrictedMode(t *testing.T) {
+	namespace := "mizu-tests"
+
+	t.Log("creating permissions for restricted user")
+	if err := ApplyKubeFilesForTest(
+		t,
+		"minikube",
+		namespace,
+		"../cli/cmd/permissionFiles/permissions-ns-tap.yaml",
+	); err != nil {
+		t.Errorf("failed to create k8s permissions, %v", err)
+		return
+	}
+
+	t.Log("switching k8s context to user")
+	if err := SwitchKubeContextForTest(t, "user-with-restricted-access"); err != nil {
+		t.Errorf("failed to switch k8s context, %v", err)
+		return
+	}
+
+	extraArgs := []string{"--set", fmt.Sprintf("mizu-resources-namespace=%s", namespace)}
+	t.Run("basic tap", func (testingT *testing.T) {basicTapTest(testingT, false, extraArgs...)})
+}

acceptanceTests/testsUtils.go

@@ -31,6 +31,7 @@ const (
 	defaultServiceName    = "httpbin"
 	defaultEntriesCount   = 50
 	waitAfterTapPodsReady = 3 * time.Second
+	allNamespaces         = ""
 )
 
 type PodDescriptor struct {
@@ -74,7 +75,7 @@ func GetApiServerUrl(port uint16) string {
 	return fmt.Sprintf("http://localhost:%v", port)
 }
 
-func GetServiceExternalIp(ctx context.Context, namespace string, service string) (string, error) {
+func NewKubernetesProvider() (*KubernetesProvider, error) {
 	home := homedir.HomeDir()
 	configLoadingRules := &clientcmd.ClientConfigLoadingRules{ExplicitPath: filepath.Join(home, ".kube", "config")}
 	clientConfig := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(
@@ -86,15 +87,23 @@ func GetServiceExternalIp(ctx context.Context, namespace string, service string)
 
 	restClientConfig, err := clientConfig.ClientConfig()
 	if err != nil {
-		return "", err
+		return nil, err
 	}
 
 	clientSet, err := kubernetes.NewForConfig(restClientConfig)
 	if err != nil {
-		return "", err
+		return nil, err
 	}
 
-	serviceObj, err := clientSet.CoreV1().Services(namespace).Get(ctx, service, metav1.GetOptions{})
+	return &KubernetesProvider{clientSet}, nil
+}
+
+type KubernetesProvider struct {
+	clientSet *kubernetes.Clientset
+}
+
+func (kp *KubernetesProvider) GetServiceExternalIp(ctx context.Context, namespace string, service string) (string, error) {
+	serviceObj, err := kp.clientSet.CoreV1().Services(namespace).Get(ctx, service, metav1.GetOptions{})
 	if err != nil {
 		return "", err
 	}
@@ -103,6 +112,105 @@ func GetServiceExternalIp(ctx context.Context, namespace string, service string)
 	return externalIp, nil
 }
 
+func SwitchKubeContextForTest(t *testing.T, newContextName string) error {
+	prevKubeContextName, err := GetKubeCurrentContextName()
+	if err != nil {
+		return err
+	}
+
+	if err := SetKubeCurrentContext(newContextName); err != nil {
+		return err
+	}
+
+	t.Cleanup(func() {
+		if err := SetKubeCurrentContext(prevKubeContextName); err != nil {
+			t.Errorf("failed to set Kubernetes context to %s, err: %v", prevKubeContextName, err)
+			t.Errorf("cleanup failed, subsequent tests may be affected")
+		}
+	})
+
+	return nil
+}
+
+func GetKubeCurrentContextName() (string, error) {
+	cmd := exec.Command("kubectl", "config", "current-context")
+
+	output, err := cmd.CombinedOutput()
+	if err != nil {
+		return "", fmt.Errorf("%v, %s", err, string(output))
+	}
+
+	return string(bytes.TrimSpace(output)), nil
+}
+
+func SetKubeCurrentContext(contextName string) error {
+	cmd := exec.Command("kubectl", "config", "use-context", contextName)
+
+	if output, err := cmd.CombinedOutput(); err != nil {
+		return fmt.Errorf("%v, %s", err, string(output))
+	}
+
+	return nil
+}
+
+func ApplyKubeFilesForTest(t *testing.T, kubeContext string, namespace string, filename ...string) error {
+	for i := range filename {
+		fname := filename[i]
+		if err := ApplyKubeFile(kubeContext, namespace, fname); err != nil {
+			return err
+		}
+
+		t.Cleanup(func() {
+			if err := DeleteKubeFile(kubeContext, namespace, fname); err != nil {
+				t.Errorf(
+					"failed to delete Kubernetes resources in namespace %s from filename %s, err: %v",
+					namespace,
+					fname,
+					err,
+				)
+			}
+		})
+	}
+
+	return nil
+}
+
+func ApplyKubeFile(kubeContext string, namespace string, filename string) (error) {
+	cmdArgs := []string{
+		"apply",
+		"--context", kubeContext,
+		"-f", filename,
+	}
+	if namespace != allNamespaces {
+		cmdArgs = append(cmdArgs, "-n", namespace)
+	}
+	cmd := exec.Command("kubectl", cmdArgs...)
+
+	if output, err := cmd.CombinedOutput(); err != nil {
+		return fmt.Errorf("%v, %s", err, string(output))
+	}
+
+	return nil
+}
+
+func DeleteKubeFile(kubeContext string, namespace string, filename string) error {
+	cmdArgs := []string{
+		"delete",
+		"--context", kubeContext,
+		"-f", filename,
+	}
+	if namespace != allNamespaces {
+		cmdArgs = append(cmdArgs, "-n", namespace)
+	}
+	cmd := exec.Command("kubectl", cmdArgs...)
+
+	if output, err := cmd.CombinedOutput(); err != nil {
+		return fmt.Errorf("%v, %s", err, string(output))
+	}
+
+	return nil
+}
+
 func getDefaultCommandArgs() []string {
 	setFlag := "--set"
 	telemetry := "telemetry=false"

agent/pkg/elastic/esClient.go

@@ -4,13 +4,14 @@ import (
 	"bytes"
 	"crypto/tls"
 	"encoding/json"
+	"net/http"
+	"sync"
+	"time"
+
 	"github.com/elastic/go-elasticsearch/v7"
 	"github.com/up9inc/mizu/shared"
 	"github.com/up9inc/mizu/shared/logger"
 	"github.com/up9inc/mizu/tap/api"
-	"net/http"
-	"sync"
-	"time"
 )
 
 type client struct {
@@ -31,6 +32,9 @@ func GetInstance() *client {
 
 func (client *client) Configure(config shared.ElasticConfig) {
 	if config.Url == "" || config.User == "" || config.Password == "" {
+		if client.es != nil {
+			client.es = nil
+		}
 		logger.Log.Infof("No elastic configuration was supplied, elastic exporter disabled")
 		return
 	}
@@ -46,13 +50,13 @@ func (client *client) Configure(config shared.ElasticConfig) {
 
 	es, err := elasticsearch.NewClient(cfg)
 	if err != nil {
-		logger.Log.Fatalf("Failed to initialize elastic client %v", err)
+		logger.Log.Errorf("Failed to initialize elastic client %v", err)
 	}
 
 	// Have the client instance return a response
 	res, err := es.Info()
 	if err != nil {
-		logger.Log.Fatalf("Elastic client.Info() ERROR: %v", err)
+		logger.Log.Errorf("Elastic client.Info() ERROR: %v", err)
 	} else {
 		client.es = es
 		client.index = "mizu_traffic_http_" + time.Now().Format("2006_01_02_15_04")

agent/pkg/oas/oas_generator.go

@@ -33,10 +33,23 @@ func (g *oasGenerator) Start() {
 	g.entriesChan = make(chan EntryWithSource, 100) // buffer up to 100 entries for OAS processing
 	g.ServiceSpecs = &sync.Map{}
 	g.started = true
-	go instance.runGeneretor()
+	go instance.runGenerator()
 }
 
-func (g *oasGenerator) runGeneretor() {
+func (g *oasGenerator) Stop() {
+	if !g.started {
+		return
+	}
+	g.cancel()
+	g.Reset()
+	g.started = false
+}
+
+func (g *oasGenerator) IsStarted() bool {
+	return g.started
+}
+
+func (g *oasGenerator) runGenerator() {
 	for {
 		select {
 		case <-g.ctx.Done():

agent/pkg/servicemap/servicemap.go

@@ -32,6 +32,7 @@ type serviceMap struct {
 
 type ServiceMap interface {
 	Enable()
+	Disable()
 	IsEnabled() bool
 	NewTCPEntry(source *tapApi.TCP, destination *tapApi.TCP, protocol *tapApi.Protocol)
 	GetStatus() ServiceMapStatus
@@ -159,6 +160,11 @@ func (s *serviceMap) Enable() {
 	s.enabled = true
 }
 
+func (s *serviceMap) Disable() {
+	s.Reset()
+	s.enabled = false
+}
+
 func (s *serviceMap) IsEnabled() bool {
 	return s.enabled
 }

cli/cmd/permissionFiles/permissions-all-namespaces-debug-optional.yaml

@@ -17,7 +17,7 @@ metadata:
   name: mizu-runner-debug-clusterrolebindings
 subjects:
 - kind: User
-  name: user1
+  name: user-with-clusterwide-access
   apiGroup: rbac.authorization.k8s.io
 roleRef:
   kind: ClusterRole

cli/cmd/permissionFiles/permissions-all-namespaces-ip-resolution-optional.yaml

@@ -29,7 +29,7 @@ metadata:
   name: mizu-resolver-clusterrolebindings
 subjects:
 - kind: User
-  name: user1
+  name: user-with-clusterwide-access
   apiGroup: rbac.authorization.k8s.io
 roleRef:
   kind: ClusterRole

cli/cmd/permissionFiles/permissions-all-namespaces-tap.yaml

@@ -22,6 +22,9 @@ rules:
 - apiGroups: [""]
   resources: ["configmaps"]
   verbs: ["create"]
+- apiGroups: [""]
+  resources: ["pods/log"]
+  verbs: ["get"]
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
@@ -29,7 +32,7 @@ metadata:
   name: mizu-runner-clusterrolebindings
 subjects:
 - kind: User
-  name: user1
+  name: user-with-clusterwide-access
   apiGroup: rbac.authorization.k8s.io
 roleRef:
   kind: ClusterRole

cli/cmd/permissionFiles/permissions-ns-debug-optional.yaml

@@ -3,7 +3,6 @@ kind: Role
 apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: mizu-runner-debug-role
-  namespace: user1
 rules:
 - apiGroups: ["events.k8s.io"]
   resources: ["events"]
@@ -16,10 +15,9 @@ kind: RoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: mizu-runner-debug-rolebindings
-  namespace: user1
 subjects:
 - kind: User
-  name: user1
+  name: user-with-restricted-access
   apiGroup: rbac.authorization.k8s.io
 roleRef:
   kind: Role

cli/cmd/permissionFiles/permissions-ns-ip-resolution-optional.yaml

@@ -3,7 +3,6 @@ kind: Role
 apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: mizu-resolver-role
-  namespace: user1
 rules:
 - apiGroups: [""]
   resources: ["serviceaccounts"]
@@ -28,10 +27,9 @@ kind: RoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: mizu-resolver-rolebindings
-  namespace: user1
 subjects:
 - kind: User
-  name: user1
+  name: user-with-restricted-access
   apiGroup: rbac.authorization.k8s.io
 roleRef:
   kind: Role

cli/cmd/permissionFiles/permissions-ns-tap.yaml

@@ -3,7 +3,6 @@ kind: Role
 apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: mizu-runner-role
-  namespace: user1
 rules:
 - apiGroups: [""]
   resources: ["pods"]
@@ -20,15 +19,17 @@ rules:
 - apiGroups: [""]
   resources: ["configmaps"]
   verbs: ["create", "delete"]
+- apiGroups: [""]
+  resources: ["pods/log"]
+  verbs: ["get"]
 ---
 kind: RoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: mizu-runner-rolebindings
-  namespace: user1
 subjects:
 - kind: User
-  name: user1
+  name: user-with-restricted-access
   apiGroup: rbac.authorization.k8s.io
 roleRef:
   kind: Role