Skip to content

Commit

Permalink
Changing Application Profile reference from pointer to object (#116)
Browse files Browse the repository at this point in the history
* Changing Application Profile refernce from pointer to object

Signed-off-by: Ben <[email protected]>

* Adding unit tests to application profile cache

Signed-off-by: Ben <[email protected]>

* fixing cache assignment

Signed-off-by: Ben <[email protected]>

* Proper cleanup in tests

Signed-off-by: Ben <[email protected]>

---------

Signed-off-by: Ben <[email protected]>
  • Loading branch information
slashben authored Dec 24, 2023
1 parent f7a0054 commit 2b0b981
Show file tree
Hide file tree
Showing 3 changed files with 195 additions and 20 deletions.
10 changes: 9 additions & 1 deletion cmd/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ import (
// Global variables
var NodeName string
var k8sConfig *rest.Config
var dynamicClientGlobal dynamic.Interface
var FinalizationDurationInSeconds int64 = 120
var FinalizationJitterInSeconds int64 = 30
var SamplingIntervalInSeconds int64 = 60
Expand Down Expand Up @@ -79,6 +80,13 @@ func serviceInitNChecks(modeEbpf bool) error {
}
k8sConfig = config

// Create Dyanmic client
dynamicClient, err := dynamic.NewForConfig(k8sConfig)
if err != nil {
return err
}
dynamicClientGlobal = dynamicClient

// Get Node name from environment variable
if nodeName := os.Getenv("NODE_NAME"); nodeName == "" {
return fmt.Errorf("NODE_NAME environment variable not set")
Expand Down Expand Up @@ -160,7 +168,7 @@ func main() {
// Create tracer (without sink for now)
tracer := tracing.NewTracer(NodeName, k8sConfig, []tracing.EventSink{}, false)
// Create application profile cache
appProfileCache, err := approfilecache.NewApplicationProfileK8sCache(k8sConfig)
appProfileCache, err := approfilecache.NewApplicationProfileK8sCache(dynamicClientGlobal)
if err != nil {
log.Fatalf("Failed to create application profile cache: %v\n", err)
}
Expand Down
34 changes: 15 additions & 19 deletions pkg/approfilecache/cache.go
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,6 @@ import (
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/client-go/dynamic"
"k8s.io/client-go/rest"
)

type ApplicationProfileCacheEntry struct {
Expand All @@ -27,10 +26,8 @@ type ApplicationProfileCacheEntry struct {
}

type ApplicationProfileK8sCache struct {
k8sConfig *rest.Config
dynamicClient *dynamic.DynamicClient

cache map[string]*ApplicationProfileCacheEntry
dynamicClient dynamic.Interface
cache map[string]ApplicationProfileCacheEntry

applicationProfileWatcher watcher.WatcherInterface

Expand All @@ -56,14 +53,9 @@ func getApplicationProfileFromUnstructured(typedObj *unstructured.Unstructured)
return &applicationProfileObj, nil
}

func NewApplicationProfileK8sCache(k8sConfig *rest.Config) (*ApplicationProfileK8sCache, error) {
dynamicClient, err := dynamic.NewForConfig(k8sConfig)
if err != nil {
return nil, err
}
cache := make(map[string]*ApplicationProfileCacheEntry)
func NewApplicationProfileK8sCache(dynamicClient dynamic.Interface) (*ApplicationProfileK8sCache, error) {
cache := make(map[string]ApplicationProfileCacheEntry)
newApplicationCache := ApplicationProfileK8sCache{
k8sConfig: k8sConfig,
dynamicClient: dynamicClient,
cache: cache,
applicationProfileWatcher: watcher.NewWatcher(dynamicClient, false), // No need to pre-list the application profiles since the container start will look for them
Expand Down Expand Up @@ -106,7 +98,7 @@ func (cache *ApplicationProfileK8sCache) LoadApplicationProfile(namespace, kind,
return fmt.Errorf("application profile %s is not final", applicationProfile.GetName())
}

cache.cache[containerID] = &ApplicationProfileCacheEntry{
cache.cache[containerID] = ApplicationProfileCacheEntry{
ApplicationProfile: applicationProfile,
WorkloadName: workloadName,
WorkloadKind: strings.ToLower(kind),
Expand All @@ -120,7 +112,7 @@ func (cache *ApplicationProfileK8sCache) LoadApplicationProfile(namespace, kind,
}

func (cache *ApplicationProfileK8sCache) AnticipateApplicationProfile(namespace, kind, workloadName, ownerKind, ownerName, containerName, containerID string, acceptPartial bool) error {
cache.cache[containerID] = &ApplicationProfileCacheEntry{
cache.cache[containerID] = ApplicationProfileCacheEntry{
ApplicationProfile: nil,
WorkloadName: workloadName,
WorkloadKind: strings.ToLower(kind),
Expand All @@ -133,7 +125,11 @@ func (cache *ApplicationProfileK8sCache) AnticipateApplicationProfile(namespace,
}

func (cache *ApplicationProfileK8sCache) DeleteApplicationProfile(containerID string) error {
delete(cache.cache, containerID)
if item, ok := cache.cache[containerID]; ok {
item.ApplicationProfile = nil
delete(cache.cache, containerID)
}

return nil
}

Expand Down Expand Up @@ -224,8 +220,6 @@ func (c *ApplicationProfileK8sCache) handleApplicationProfile(appProfileUnstruct
partial := appProfileUnstructured.GetLabels()["kapprofiler.kubescape.io/partial"] == "true"
final := appProfileUnstructured.GetLabels()["kapprofiler.kubescape.io/final"] == "true"

log.Printf("APCache: Handling application profile %s/%s, partial: %v, final: %v\n", appProfileUnstructured.GetNamespace(), appProfileUnstructured.GetName(), partial, final)

// Check if the application profile is final or partial, if not then skip it
if !final {
return
Expand All @@ -237,7 +231,7 @@ func (c *ApplicationProfileK8sCache) handleApplicationProfile(appProfileUnstruct
// Add the application profile to the cache

// Loop over the application profile cache entries and check if there is an entry for the same workload
for _, cacheEntry := range c.cache {
for id, cacheEntry := range c.cache {
if cacheEntry.Namespace == appProfileUnstructured.GetNamespace() {
if !cacheEntry.AcceptPartial && partial {
// Skip the partial application profile becuase we expect a final one
Expand All @@ -251,9 +245,11 @@ func (c *ApplicationProfileK8sCache) handleApplicationProfile(appProfileUnstruct
return
}

log.Printf("Updating application profile %s for workload %s/%s\n", appProfile.GetName(), cacheEntry.Namespace, cacheEntry.WorkloadName)

// Update the cache entry
cacheEntry.ApplicationProfile = appProfile
cacheEntry.OwnerLevelProfile = false
c.cache[id] = cacheEntry
continue
}
}
Expand Down
171 changes: 171 additions & 0 deletions pkg/approfilecache/cache_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,171 @@
package approfilecache

import (
"context"
"testing"
"time"

"github.com/kubescape/kapprofiler/pkg/collector"
v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
dfake "k8s.io/client-go/dynamic/fake"
)

func TestCacheBasicExists(t *testing.T) {
// ApplicationProfile
appProfile := collector.ApplicationProfile{
ObjectMeta: v1.ObjectMeta{
Name: "deployment-nginx",
Namespace: "default",
Labels: map[string]string{
"kapprofiler.kubescape.io/final": "true",
},
},
Spec: collector.ApplicationProfileSpec{
Containers: []collector.ContainerProfile{
{
Name: "nginx",
Execs: []collector.ExecCalls{
{
Path: "/bin/bash",
Args: []string{"-c", "echo hello"},
Envs: []string{"PATH=/bin"},
},
},
},
},
},
}

// Convert application profile to unstructured
appProfileUnstructured, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&appProfile)
if err != nil {
t.Errorf("Failed to convert application profile to unstructured: %v", err)
return
}

// Setup your test if needed
dynamicClient := dfake.NewSimpleDynamicClientWithCustomListKinds(runtime.NewScheme(), map[schema.GroupVersionResource]string{
collector.AppProfileGvr: collector.ApplicationProfileKind + "List",
schema.GroupVersionResource{
Group: "",
Version: "v1",
Resource: "pods",
}: "PodList",
schema.GroupVersionResource{
Group: "",
Version: "v1",
Resource: "namespaces",
}: "NamespaceList",
})

// Add the ApplicationProfile to the fake dynamic client
_, err = dynamicClient.Resource(collector.AppProfileGvr).Namespace("default").Create(context.Background(), &unstructured.Unstructured{Object: appProfileUnstructured}, v1.CreateOptions{})
if err != nil {
t.Errorf("Failed to create application profile: %v", err)
return
}

cache, err := NewApplicationProfileK8sCache(dynamicClient)
if err != nil {
t.Errorf("Failed to create cache: %v", err)
return
}
defer cache.Destroy()

// Load a container profile
err = cache.LoadApplicationProfile("default", "pod", "nginx-aaaaa-bbbb", "deployment", "nginx", "nginx", "00000000000000000000000000000000", false)
if err != nil {
t.Errorf("Failed to load container profile: %v", err)
return
}

// Check if the container profile is in the cache
_, err = cache.GetApplicationProfileAccess("nginx", "00000000000000000000000000000000")
if err != nil {
t.Errorf("Failed to get container profile: %v", err)
return
}
}

func TestCacheBasicAnticipateProfile(t *testing.T) {
// ApplicationProfile
appProfile := collector.ApplicationProfile{
ObjectMeta: v1.ObjectMeta{
Name: "deployment-nginx",
Namespace: "default",
Labels: map[string]string{
"kapprofiler.kubescape.io/final": "true",
},
},
Spec: collector.ApplicationProfileSpec{
Containers: []collector.ContainerProfile{
{
Name: "nginx",
Execs: []collector.ExecCalls{
{
Path: "/bin/bash",
Args: []string{"-c", "echo hello"},
Envs: []string{"PATH=/bin"},
},
},
},
},
},
}

// Convert application profile to unstructured
appProfileUnstructured, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&appProfile)
if err != nil {
t.Errorf("Failed to convert application profile to unstructured: %v", err)
return
}

// Setup your test if needed
dynamicClient := dfake.NewSimpleDynamicClientWithCustomListKinds(runtime.NewScheme(), map[schema.GroupVersionResource]string{
collector.AppProfileGvr: collector.ApplicationProfileKind + "List",
schema.GroupVersionResource{
Group: "",
Version: "v1",
Resource: "pods",
}: "PodList",
schema.GroupVersionResource{
Group: "",
Version: "v1",
Resource: "namespaces",
}: "NamespaceList",
})

cache, err := NewApplicationProfileK8sCache(dynamicClient)
if err != nil {
t.Errorf("Failed to create cache: %v", err)
return
}
defer cache.Destroy()

// Create Anticipation a container profile
err = cache.AnticipateApplicationProfile("default", "pod", "nginx-aaaaa-bbbb", "deployment", "nginx", "nginx", "00000000000000000000000000000000", false)
if err != nil {
t.Errorf("Failed to anticipate container profile: %v", err)
return
}

// Add the ApplicationProfile to the fake dynamic client
_, err = dynamicClient.Resource(collector.AppProfileGvr).Namespace("default").Create(context.Background(), &unstructured.Unstructured{Object: appProfileUnstructured}, v1.CreateOptions{})
if err != nil {
t.Errorf("Failed to create application profile: %v", err)
return
}

// Wait a second for the cache to be updated
time.Sleep(1 * time.Second)

// Check if the container profile is in the cache
_, err = cache.GetApplicationProfileAccess("nginx", "00000000000000000000000000000000")
if err != nil {
t.Errorf("Failed to get container profile: %v", err)
return
}
}

0 comments on commit 2b0b981

Please sign in to comment.