github/kubeshark
1
0
Fork 0
mirror of https://github.com/kubeshark/kubeshark.git synced 2026-07-10 00:51:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

Update go.mod, main.go, and 4 more files...

Browse Source
This commit is contained in:
RamiBerm
2021-05-10 15:44:38 +03:00
parent c4d44ba9d3
commit 4429de93b6
6 changed files with 146 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
cli/Makefile
View File

@@ -1,5 +1,5 @@
install:
go build install mizu.go
go install mizu.go
build:
go build -o bin/mizu mizu.go
@@ -16,6 +16,6 @@ build-cr:
@#GOOS=linux GOARCH=arm64 go build -o bin/mizu-linux-arm64 mizu.go
@#GOOS=windows GOARCH=arm64 go build -o bin/mizu-windows-arm64 mizu.go
clean:
clean:
#go clean
rm -f ./bin/*

5
cli/cmd/version.go
View File

@@ -2,17 +2,16 @@ package cmd
import (
"fmt"
"github.com/up9inc/mizu/cli/mizu"
"github.com/spf13/cobra"
)
var Version = "0.1.0"
var versionCmd = &cobra.Command{
Use: "version",
Short: "Print version info",
RunE: func(cmd *cobra.Command, args []string) error {
fmt.Printf("mizu version %s\n", Version)
fmt.Printf("mizu version %s\n", mizu.Version)
return nil
},
}

94
cli/kubernetes/provider.go
View File

@@ -3,10 +3,11 @@ package kubernetes
import (
_ "bytes"
"context"
"errors"
"fmt"
"strings"
core "k8s.io/api/core/v1"
rbac "k8s.io/api/rbac/v1"
k8serrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/watch"
"k8s.io/client-go/kubernetes"
@@ -19,6 +20,7 @@ import (
_ "k8s.io/client-go/tools/portforward"
"k8s.io/client-go/util/homedir"
"path/filepath"
"strings"
)
type Provider struct {
@@ -28,6 +30,11 @@ type Provider struct {
Namespace string
}
const (
serviceAccountName = "mizu-service-account"
MizuResourcesNamespace = "default"
)
func NewProvider(kubeConfigPath string, overrideNamespace string) *Provider {
kubernetesConfig := loadKubernetesConfiguration(kubeConfigPath)
restClientConfig, err := kubernetesConfig.ClientConfig()
@@ -55,8 +62,8 @@ func NewProvider(kubeConfigPath string, overrideNamespace string) *Provider {
}
}
func (provider *Provider) GetPodWatcher(ctx context.Context) watch.Interface {
watcher, err := provider.clientSet.CoreV1().Pods(provider.Namespace).Watch(ctx, metav1.ListOptions{Watch: true})
func (provider *Provider) GetPodWatcher(ctx context.Context, namespace string) watch.Interface {
watcher, err := provider.clientSet.CoreV1().Pods(namespace).Watch(ctx, metav1.ListOptions{Watch: true})
if err != nil {
panic(err.Error())
}
@@ -87,7 +94,7 @@ func (provider *Provider) CreateMizuPod(ctx context.Context, podName string, pod
pod := &core.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: podName,
Namespace: provider.Namespace,
Namespace: MizuResourcesNamespace,
},
Spec: core.PodSpec{
HostNetwork: true, // very important to make passive tapper see traffic
@@ -111,15 +118,88 @@ func (provider *Provider) CreateMizuPod(ctx context.Context, podName string, pod
},
},
},
ServiceAccountName: serviceAccountName,
TerminationGracePeriodSeconds: new(int64),
NodeSelector: map[string]string{"kubernetes.io/hostname": tappedPod.Spec.NodeName},
},
}
return provider.clientSet.CoreV1().Pods(provider.Namespace).Create(ctx, pod, metav1.CreateOptions{})
return provider.clientSet.CoreV1().Pods(MizuResourcesNamespace).Create(ctx, pod, metav1.CreateOptions{})
}
func (provider *Provider) DoesMizuRBACExist(ctx context.Context) (bool, error){
serviceAccount, err := provider.clientSet.CoreV1().ServiceAccounts(MizuResourcesNamespace).Get(ctx, serviceAccountName, metav1.GetOptions{})
var statusError *k8serrors.StatusError
if errors.As(err, &statusError) {
// expected behavior when resource does not exist
if statusError.ErrStatus.Reason == metav1.StatusReasonNotFound {
return false, nil
}
}
if err != nil {
return false, err
}
return serviceAccount != nil, nil
}
func (provider *Provider) CreateMizuRBAC(ctx context.Context, version string) error {
clusterRoleName := "mizu-cluster-role"
serviceAccount := &core.ServiceAccount{
ObjectMeta: metav1.ObjectMeta{
Name: serviceAccountName,
Namespace: MizuResourcesNamespace,
Labels: map[string]string{"mizu-cli-version": version},
},
}
clusterRole := &rbac.ClusterRole{
ObjectMeta: metav1.ObjectMeta{
Name: clusterRoleName,
Labels: map[string]string{"mizu-cli-version": version},
},
Rules: []rbac.PolicyRule{
{
APIGroups: []string {"", "extensions", "apps"},
Resources: []string {"pods", "services", "endpoints"},
Verbs: []string {"list", "get", "watch"},
},
},
}
clusterRoleBinding := &rbac.ClusterRoleBinding{
ObjectMeta: metav1.ObjectMeta{
Name: "mizu-cluster-role-binding",
Labels: map[string]string{"mizu-cli-version": version},
},
RoleRef: rbac.RoleRef{
Name: clusterRoleName,
Kind: "ClusterRole",
APIGroup: "rbac.authorization.k8s.io",
},
Subjects: []rbac.Subject{
{
Kind: "ServiceAccount",
Name: serviceAccountName,
Namespace: MizuResourcesNamespace,
},
},
}
_, err := provider.clientSet.CoreV1().ServiceAccounts(MizuResourcesNamespace).Create(ctx, serviceAccount, metav1.CreateOptions{})
if err != nil {
return err
}
_, err = provider.clientSet.RbacV1().ClusterRoles().Create(ctx, clusterRole, metav1.CreateOptions{})
if err != nil {
return err
}
_, err = provider.clientSet.RbacV1().ClusterRoleBindings().Create(ctx, clusterRoleBinding, metav1.CreateOptions{})
if err != nil {
return err
}
return nil
}
func (provider *Provider) RemovePod(ctx context.Context, podName string) {
provider.clientSet.CoreV1().Pods(provider.Namespace).Delete(ctx, podName, metav1.DeleteOptions{})
provider.clientSet.CoreV1().Pods(MizuResourcesNamespace).Delete(ctx, podName, metav1.DeleteOptions{})
}
func getClientSet(config *restclient.Config) *kubernetes.Clientset {

30
cli/mizu/mizuRunner.go
View File

@@ -12,6 +12,8 @@ import (
"time"
)
const Version = "0.1.0"
func Run(tappedPodName string) {
kubernetesProvider := kubernetes.NewProvider(config.Configuration.KubeConfigPath, config.Configuration.Namespace)
ctx, cancel := context.WithCancel(context.Background())
@@ -19,7 +21,8 @@ func Run(tappedPodName string) {
podName := "mizu-collector"
go createPodAndPortForward(ctx, kubernetesProvider, cancel, podName, tappedPodName) //TODO convert this to job for built in pod ttl or have the running app handle this
createRBACIfNecessary(ctx, kubernetesProvider, cancel)
go createPodAndPortForward(ctx, kubernetesProvider, cancel, podName, kubernetes.MizuResourcesNamespace, tappedPodName) //TODO convert this to job for built in pod ttl or have the running app handle this
waitForFinish(ctx, cancel) //block until exit signal or error
// TODO handle incoming traffic from tapper using a channel
@@ -31,7 +34,7 @@ func Run(tappedPodName string) {
}
func watchPodsForTapping(ctx context.Context, kubernetesProvider *kubernetes.Provider, cancel context.CancelFunc, podRegex *regexp.Regexp) {
added, modified, removed, errorChan := kubernetes.FilteredWatch(ctx, kubernetesProvider.GetPodWatcher(ctx), podRegex)
added, modified, removed, errorChan := kubernetes.FilteredWatch(ctx, kubernetesProvider.GetPodWatcher(ctx, kubernetesProvider.Namespace), podRegex)
for {
select {
case newTarget := <- added:
@@ -52,7 +55,7 @@ func watchPodsForTapping(ctx context.Context, kubernetesProvider *kubernetes.Pro
}
}
func createPodAndPortForward(ctx context.Context, kubernetesProvider *kubernetes.Provider, cancel context.CancelFunc, podName string, tappedPodName string) {
func createPodAndPortForward(ctx context.Context, kubernetesProvider *kubernetes.Provider, cancel context.CancelFunc, podName string, namespace string, tappedPodName string) {
pod, err := kubernetesProvider.CreateMizuPod(ctx, podName, config.Configuration.MizuImage, tappedPodName)
if err != nil {
fmt.Printf("error creating pod %s", err)
@@ -60,7 +63,7 @@ func createPodAndPortForward(ctx context.Context, kubernetesProvider *kubernetes
return
}
podExactRegex := regexp.MustCompile(fmt.Sprintf("^%s$", pod.Name))
added, modified, removed, errorChan := kubernetes.FilteredWatch(ctx, kubernetesProvider.GetPodWatcher(ctx), podExactRegex)
added, modified, removed, errorChan := kubernetes.FilteredWatch(ctx, kubernetesProvider.GetPodWatcher(ctx, namespace), podExactRegex)
isPodReady := false
var portForward *kubernetes.PortForward
for {
@@ -75,7 +78,7 @@ func createPodAndPortForward(ctx context.Context, kubernetesProvider *kubernetes
if modifiedPod.Status.Phase == "Running" && !isPodReady {
isPodReady = true
var err error
portForward, err = kubernetes.NewPortForward(kubernetesProvider, kubernetesProvider.Namespace, podName, config.Configuration.GuiPort, config.Configuration.MizuPodPort, cancel)
portForward, err = kubernetes.NewPortForward(kubernetesProvider, namespace, podName, config.Configuration.GuiPort, config.Configuration.MizuPodPort, cancel)
fmt.Printf("Web interface is now available at http://localhost:%d\n", config.Configuration.GuiPort)
if err != nil {
fmt.Printf("error forwarding port to pod %s\n", err)
@@ -101,6 +104,23 @@ func createPodAndPortForward(ctx context.Context, kubernetesProvider *kubernetes
}
}
func createRBACIfNecessary(ctx context.Context, kubernetesProvider *kubernetes.Provider, cancel context.CancelFunc) {
mizuRBACExists, err := kubernetesProvider.DoesMizuRBACExist(ctx)
if err != nil {
fmt.Printf("error checking rbac %v", err)
cancel()
return
}
if !mizuRBACExists {
err := kubernetesProvider.CreateMizuRBAC(ctx, Version)
if err != nil {
fmt.Printf("error creating rbac %v", err)
cancel()
return
}
}
}
func waitForFinish(ctx context.Context, cancel context.CancelFunc) {
sigChan := make(chan os.Signal, 1)
signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)