github/kubeshark
1
0
Fork 0
mirror of https://github.com/kubeshark/kubeshark.git synced 2025-09-08 14:01:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fixes (#171)

Browse Source
This commit is contained in:
Igor Gov
2021-08-05 19:29:06 +03:00
committed by GitHub
parent 60533a9591
commit 0244f12167
16 changed files with 245 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
agent/pkg/resolver/loader.go
View File

@@ -1,6 +1,7 @@
package resolver
import (
cmap "github.com/orcaman/concurrent-map"
"k8s.io/client-go/kubernetes"
_ "k8s.io/client-go/plugin/pkg/client/auth/azure"
_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
@@ -9,7 +10,7 @@ import (
restclient "k8s.io/client-go/rest"
)
func NewFromInCluster(errOut chan error) (*Resolver, error) {
func NewFromInCluster(errOut chan error, namesapce string) (*Resolver, error) {
config, err := restclient.InClusterConfig()
if err != nil {
return nil, err
@@ -18,5 +19,5 @@ func NewFromInCluster(errOut chan error) (*Resolver, error) {
if err != nil {
return nil, err
}
return &Resolver{clientConfig: config, clientSet: clientset, nameMap: make(map[string]string), serviceMap: make(map[string]string), errOut: errOut}, nil
return &Resolver{clientConfig: config, clientSet: clientset, nameMap: cmap.New(), serviceMap: cmap.New(), errOut: errOut, namespace: namesapce}, nil
}

106
agent/pkg/resolver/resolver.go
View File

@@ -7,12 +7,14 @@ import (
"github.com/romana/rlog"
k8serrors "k8s.io/apimachinery/pkg/api/errors"
"github.com/orcaman/concurrent-map"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/watch"
"k8s.io/client-go/kubernetes"
restclient "k8s.io/client-go/rest"
)
const (
kubClientNullString = "None"
)
@@ -20,17 +22,16 @@ const (
type Resolver struct {
clientConfig *restclient.Config
clientSet *kubernetes.Clientset
nameMap map[string]string
serviceMap map[string]string
nameMap cmap.ConcurrentMap
serviceMap cmap.ConcurrentMap
isStarted bool
errOut chan error
namespace string
}
func (resolver *Resolver) Start(ctx context.Context, namespace string) {
func (resolver *Resolver) Start(ctx context.Context) {
if !resolver.isStarted {
resolver.isStarted = true
resolver.namespace = namespace
go resolver.infiniteErrorHandleRetryFunc(ctx, resolver.watchServices)
go resolver.infiniteErrorHandleRetryFunc(ctx, resolver.watchEndpoints)
@@ -39,19 +40,19 @@ func (resolver *Resolver) Start(ctx context.Context, namespace string) {
}
func (resolver *Resolver) Resolve(name string) string {
resolvedName, isFound := resolver.nameMap[name]
resolvedName, isFound := resolver.nameMap.Get(name)
if !isFound {
return ""
}
return resolvedName
return resolvedName.(string)
}
func (resolver *Resolver) GetMap() map[string]string {
func (resolver *Resolver) GetMap() cmap.ConcurrentMap {
return resolver.nameMap
}
func (resolver *Resolver) CheckIsServiceIP(address string) bool {
_, isFound := resolver.serviceMap[address]
_, isFound := resolver.serviceMap.Get(address)
return isFound
}
@@ -63,17 +64,17 @@ func (resolver *Resolver) watchPods(ctx context.Context) error {
}
for {
select {
case event := <- watcher.ResultChan():
if event.Object == nil {
return errors.New("error in kubectl pod watch")
}
if event.Type == watch.Deleted {
pod := event.Object.(*corev1.Pod)
resolver.saveResolvedName(pod.Status.PodIP, "", event.Type)
}
case <- ctx.Done():
watcher.Stop()
return nil
case event := <-watcher.ResultChan():
if event.Object == nil {
return errors.New("error in kubectl pod watch")
}
if event.Type == watch.Deleted {
pod := event.Object.(*corev1.Pod)
resolver.saveResolvedName(pod.Status.PodIP, "", event.Type)
}
case <-ctx.Done():
watcher.Stop()
return nil
}
}
}
@@ -86,37 +87,37 @@ func (resolver *Resolver) watchEndpoints(ctx context.Context) error {
}
for {
select {
case event := <- watcher.ResultChan():
if event.Object == nil {
return errors.New("error in kubectl endpoint watch")
}
endpoint := event.Object.(*corev1.Endpoints)
serviceHostname := fmt.Sprintf("%s.%s", endpoint.Name, endpoint.Namespace)
if endpoint.Subsets != nil {
for _, subset := range endpoint.Subsets {
var ports []int32
if subset.Ports != nil {
for _, portMapping := range subset.Ports {
if portMapping.Port > 0 {
ports = append(ports, portMapping.Port)
}
case event := <-watcher.ResultChan():
if event.Object == nil {
return errors.New("error in kubectl endpoint watch")
}
endpoint := event.Object.(*corev1.Endpoints)
serviceHostname := fmt.Sprintf("%s.%s", endpoint.Name, endpoint.Namespace)
if endpoint.Subsets != nil {
for _, subset := range endpoint.Subsets {
var ports []int32
if subset.Ports != nil {
for _, portMapping := range subset.Ports {
if portMapping.Port > 0 {
ports = append(ports, portMapping.Port)
}
}
if subset.Addresses != nil {
for _, address := range subset.Addresses {
resolver.saveResolvedName(address.IP, serviceHostname, event.Type)
for _, port := range ports {
ipWithPort := fmt.Sprintf("%s:%d", address.IP, port)
resolver.saveResolvedName(ipWithPort, serviceHostname, event.Type)
}
}
}
}
if subset.Addresses != nil {
for _, address := range subset.Addresses {
resolver.saveResolvedName(address.IP, serviceHostname, event.Type)
for _, port := range ports {
ipWithPort := fmt.Sprintf("%s:%d", address.IP, port)
resolver.saveResolvedName(ipWithPort, serviceHostname, event.Type)
}
}
}
}
case <- ctx.Done():
watcher.Stop()
return nil
}
case <-ctx.Done():
watcher.Stop()
return nil
}
}
}
@@ -129,7 +130,7 @@ func (resolver *Resolver) watchServices(ctx context.Context) error {
}
for {
select {
case event := <- watcher.ResultChan():
case event := <-watcher.ResultChan():
if event.Object == nil {
return errors.New("error in kubectl service watch")
}
@@ -145,7 +146,7 @@ func (resolver *Resolver) watchServices(ctx context.Context) error {
resolver.saveResolvedName(ingress.IP, serviceHostname, event.Type)
}
}
case <- ctx.Done():
case <-ctx.Done():
watcher.Stop()
return nil
}
@@ -154,19 +155,19 @@ func (resolver *Resolver) watchServices(ctx context.Context) error {
func (resolver *Resolver) saveResolvedName(key string, resolved string, eventType watch.EventType) {
if eventType == watch.Deleted {
delete(resolver.nameMap, key)
resolver.nameMap.Remove(key)
rlog.Infof("setting %s=nil\n", key)
} else {
resolver.nameMap[key] = resolved
resolver.nameMap.Set(key, resolved)
rlog.Infof("setting %s=%s\n", key, resolved)
}
}
func (resolver *Resolver) saveServiceIP(key string, resolved string, eventType watch.EventType) {
if eventType == watch.Deleted {
delete(resolver.serviceMap, key)
resolver.serviceMap.Remove(key)
} else {
resolver.serviceMap[key] = resolved
resolver.serviceMap.Set(key, resolved)
}
}
@@ -189,4 +190,3 @@ func (resolver *Resolver) infiniteErrorHandleRetryFunc(ctx context.Context, fun
}
}
}