github/kubeshark
1
0
Fork 0
mirror of https://github.com/kubeshark/kubeshark.git synced 2025-09-13 05:11:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Technical depth: Adding Go linter to CI (#734)

Browse Source
This commit is contained in:
Igor Gov
2022-02-01 08:47:26 +02:00
committed by GitHub
parent daf6b3db06
commit 0f6c56986f
37 changed files with 203 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
shared/kubernetes/eventWatchHelper.go
View File

@@ -34,7 +34,7 @@ func (wh *EventWatchHelper) Filter(wEvent *WatchEvent) (bool, error) {
return false, nil
}
if strings.ToLower(event.Regarding.Kind) != strings.ToLower(wh.Kind) {
if strings.EqualFold(event.Regarding.Kind, wh.Kind) {
return false, nil
}

25
shared/kubernetes/mizuTapperSyncer.go
View File

@@ -3,21 +3,22 @@ package kubernetes
import (
"context"
"fmt"
"regexp"
"time"
"github.com/op/go-logging"
"github.com/up9inc/mizu/shared"
"github.com/up9inc/mizu/shared/debounce"
"github.com/up9inc/mizu/shared/logger"
"github.com/up9inc/mizu/tap/api"
core "k8s.io/api/core/v1"
"regexp"
"time"
)
const updateTappersDelay = 5 * time.Second
type TappedPodChangeEvent struct {
Added []core.Pod
Removed []core.Pod
Added []core.Pod
Removed []core.Pod
}
// MizuTapperSyncer uses a k8s pod watch to update tapper daemonsets when targeted pods are removed or created
@@ -222,10 +223,14 @@ func (tapperSyncer *MizuTapperSyncer) watchPodsForTapping() {
switch wEvent.Type {
case EventAdded:
logger.Log.Debugf("Added matching pod %s, ns: %s", pod.Name, pod.Namespace)
restartTappersDebouncer.SetOn()
if err := restartTappersDebouncer.SetOn(); err != nil {
logger.Log.Error(err)
}
case EventDeleted:
logger.Log.Debugf("Removed matching pod %s, ns: %s", pod.Name, pod.Namespace)
restartTappersDebouncer.SetOn()
if err := restartTappersDebouncer.SetOn(); err != nil {
logger.Log.Error(err)
}
case EventModified:
logger.Log.Debugf("Modified matching pod %s, ns: %s, phase: %s, ip: %s", pod.Name, pod.Namespace, pod.Status.Phase, pod.Status.PodIP)
// Act only if the modified pod has already obtained an IP address.
@@ -235,7 +240,9 @@ func (tapperSyncer *MizuTapperSyncer) watchPodsForTapping() {
// - Pod reaches ready state
// Ready/unready transitions might also trigger this event.
if pod.Status.PodIP != "" {
restartTappersDebouncer.SetOn()
if err := restartTappersDebouncer.SetOn(); err != nil {
logger.Log.Error(err)
}
}
case EventBookmark:
break
@@ -285,8 +292,8 @@ func (tapperSyncer *MizuTapperSyncer) updateCurrentlyTappedPods() (err error, ch
tapperSyncer.CurrentlyTappedPods = podsToTap
tapperSyncer.nodeToTappedPodMap = GetNodeHostToTappedPodsMap(tapperSyncer.CurrentlyTappedPods)
tapperSyncer.TapPodChangesOut <- TappedPodChangeEvent{
Added: addedPods,
Removed: removedPods,
Added: addedPods,
Removed: removedPods,
}
return nil, true
}

18
shared/kubernetes/provider.go
View File

@@ -194,19 +194,19 @@ func (provider *Provider) GetMizuApiServerPodObject(opts *ApiServerOptions, moun
cpuLimit, err := resource.ParseQuantity(opts.Resources.CpuLimit)
if err != nil {
return nil, errors.New(fmt.Sprintf("invalid cpu limit for %s container", opts.PodName))
return nil, fmt.Errorf("invalid cpu limit for %s container", opts.PodName)
}
memLimit, err := resource.ParseQuantity(opts.Resources.MemoryLimit)
if err != nil {
return nil, errors.New(fmt.Sprintf("invalid memory limit for %s container", opts.PodName))
return nil, fmt.Errorf("invalid memory limit for %s container", opts.PodName)
}
cpuRequests, err := resource.ParseQuantity(opts.Resources.CpuRequests)
if err != nil {
return nil, errors.New(fmt.Sprintf("invalid cpu request for %s container", opts.PodName))
return nil, fmt.Errorf("invalid cpu request for %s container", opts.PodName)
}
memRequests, err := resource.ParseQuantity(opts.Resources.MemoryRequests)
if err != nil {
return nil, errors.New(fmt.Sprintf("invalid memory request for %s container", opts.PodName))
return nil, fmt.Errorf("invalid memory request for %s container", opts.PodName)
}
command := []string{"./mizuagent", "--api-server"}
@@ -395,7 +395,7 @@ func (provider *Provider) CreatePod(ctx context.Context, namespace string, podSp
}
func (provider *Provider) CreateDeployment(ctx context.Context, namespace string, deploymentName string, podSpec *core.Pod) (*v1.Deployment, error) {
if _, keyExists := podSpec.ObjectMeta.Labels["app"]; keyExists == false {
if _, keyExists := podSpec.ObjectMeta.Labels["app"]; !keyExists {
return nil, errors.New("pod spec must contain 'app' label")
}
podTemplate := &core.PodTemplateSpec{
@@ -854,19 +854,19 @@ func (provider *Provider) ApplyMizuTapperDaemonSet(ctx context.Context, namespac
)
cpuLimit, err := resource.ParseQuantity(resources.CpuLimit)
if err != nil {
return errors.New(fmt.Sprintf("invalid cpu limit for %s container", tapperPodName))
return fmt.Errorf("invalid cpu limit for %s container", tapperPodName)
}
memLimit, err := resource.ParseQuantity(resources.MemoryLimit)
if err != nil {
return errors.New(fmt.Sprintf("invalid memory limit for %s container", tapperPodName))
return fmt.Errorf("invalid memory limit for %s container", tapperPodName)
}
cpuRequests, err := resource.ParseQuantity(resources.CpuRequests)
if err != nil {
return errors.New(fmt.Sprintf("invalid cpu request for %s container", tapperPodName))
return fmt.Errorf("invalid cpu request for %s container", tapperPodName)
}
memRequests, err := resource.ParseQuantity(resources.MemoryRequests)
if err != nil {
return errors.New(fmt.Sprintf("invalid memory request for %s container", tapperPodName))
return fmt.Errorf("invalid memory request for %s container", tapperPodName)
}
agentResourceLimits := core.ResourceList{
"cpu": cpuLimit,

13
shared/kubernetes/watch.go
View File

@@ -4,11 +4,12 @@ import (
"context"
"errors"
"fmt"
"github.com/up9inc/mizu/shared/debounce"
"github.com/up9inc/mizu/shared/logger"
"sync"
"time"
"github.com/up9inc/mizu/shared/debounce"
"github.com/up9inc/mizu/shared/logger"
"k8s.io/apimachinery/pkg/watch"
)
@@ -31,7 +32,7 @@ func FilteredWatch(ctx context.Context, watcherCreator WatchCreator, targetNames
go func(targetNamespace string) {
defer wg.Done()
watchRestartDebouncer := debounce.NewDebouncer(1 * time.Minute, func() {})
watchRestartDebouncer := debounce.NewDebouncer(1*time.Minute, func() {})
for {
watcher, err := watcherCreator.NewWatcher(ctx, targetNamespace)
@@ -44,7 +45,7 @@ func FilteredWatch(ctx context.Context, watcherCreator WatchCreator, targetNames
watcher.Stop()
select {
case <- ctx.Done():
case <-ctx.Done():
return
default:
break
@@ -55,7 +56,9 @@ func FilteredWatch(ctx context.Context, watcherCreator WatchCreator, targetNames
break
} else {
if !watchRestartDebouncer.IsOn() {
watchRestartDebouncer.SetOn()
if err := watchRestartDebouncer.SetOn(); err != nil {
logger.Log.Error(err)
}
logger.Log.Debug("k8s watch channel closed, restarting watcher")
time.Sleep(time.Second * 5)
continue