Refactor and enrich cloudbeat events

go.mod

@@ -53,7 +53,7 @@ require (
 	github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21 // indirect
 	github.com/eapache/queue v1.1.0 // indirect
 	github.com/elastic/beats/v7 v7.0.0-alpha2.0.20220413140705-d101ba1d2ae5
-	github.com/elastic/csp-security-policies v0.0.14-go-lib
+	github.com/elastic/csp-security-policies v0.0.15-go-lib
 	github.com/ghodss/yaml v1.0.0 // indirect
 	github.com/go-ini/ini v1.66.4 // indirect
 	github.com/go-logr/logr v1.2.2 // indirect

go.sum

@@ -517,6 +517,10 @@ github.com/elastic/beats/v7 v7.0.0-alpha2.0.20220413140705-d101ba1d2ae5 h1:kINUJ
 github.com/elastic/beats/v7 v7.0.0-alpha2.0.20220413140705-d101ba1d2ae5/go.mod h1:PhsCH91qJN33+rN/L8q2jWILmswlezJ6T+MMM6EDc8g=
 github.com/elastic/csp-security-policies v0.0.14-go-lib h1:B/Ylu4aLhfcOcM3YivzwMd4BB/NoT/jFIV98fXtZDVA=
 github.com/elastic/csp-security-policies v0.0.14-go-lib/go.mod h1:24NNr0b/5HTGtndJOmhrefb59rd7NjuqI/To39tgn+w=
+github.com/elastic/csp-security-policies v0.0.15-go-lib h1:lLmtmmL9Gngo8F8y0K1dwA6UW00sfIW3zq4t8CjuHkU=
+github.com/elastic/csp-security-policies v0.0.15-go-lib/go.mod h1:24NNr0b/5HTGtndJOmhrefb59rd7NjuqI/To39tgn+w=
+github.com/elastic/csp-security-policies v0.0.15-go-lib-alpha h1:hiVX1rm/jnBt7eiSQkDo8xTx6mF8kyg1kQpnnc21zpI=
+github.com/elastic/csp-security-policies v0.0.15-go-lib-alpha/go.mod h1:24NNr0b/5HTGtndJOmhrefb59rd7NjuqI/To39tgn+w=
 github.com/elastic/elastic-agent-client/v7 v7.0.0-20210727140539-f0905d9377f6 h1:nFvXHBjYK3e9+xF0WKDeAKK4aOO51uC28s+L9rBmilo=
 github.com/elastic/elastic-agent-client/v7 v7.0.0-20210727140539-f0905d9377f6/go.mod h1:uh/Gj9a0XEbYoM4NYz4LvaBVARz3QXLmlNjsrKY9fTc=
 github.com/elastic/elastic-agent-libs v0.1.1/go.mod h1:h8K/f7RcdxM2f19VahcS1jeco170ItqV9N7HyYsn9Ss=

resources/fetchers/ecr_fetcher.go

@@ -94,10 +94,16 @@ func (f *ECRFetcher) getAwsPodRepositories(ctx context.Context) ([]string, error
 	return repositories, nil
 }
 
-// GetID TODO: Add resource id logic to all AWS resources
-func (res ECRResource) GetID() (string, error) {
-	return "", nil
-}
 func (res ECRResource) GetData() interface{} {
 	return res
 }
+
+func (res ECRResource) GetMetadata() fetching.ResourceMetadata {
+	//TODO implement me
+	return fetching.ResourceMetadata{
+		ID:      "",
+		Type:    "",
+		SubType: "",
+		Name:    "",
+	}
+}

resources/fetchers/eks_fetcher.go

@@ -54,11 +54,16 @@ func (f EKSFetcher) Fetch(ctx context.Context) ([]fetching.Resource, error) {
 func (f EKSFetcher) Stop() {
 }
 
-// GetID TODO: Add resource id logic to all AWS resources
-func (r EKSResource) GetID() (string, error) {
-	return "", nil
-}
-
 func (r EKSResource) GetData() interface{} {
 	return r
 }
+
+func (r EKSResource) GetMetadata() fetching.ResourceMetadata {
+	//TODO implement me
+	return fetching.ResourceMetadata{
+		ID:      "",
+		Type:    "",
+		SubType: "",
+		Name:    "",
+	}
+}

resources/fetchers/elb_fetcher.go

@@ -92,11 +92,16 @@ func (f *ELBFetcher) GetLoadBalancers() ([]string, error) {
 func (f *ELBFetcher) Stop() {
 }
 
-// GetID TODO: Add resource id logic to all AWS resources
-func (r ELBResource) GetID() (string, error) {
-	return "", nil
-}
-
 func (r ELBResource) GetData() interface{} {
 	return r
 }
+
+func (r ELBResource) GetMetadata() fetching.ResourceMetadata {
+	//TODO implement me
+	return fetching.ResourceMetadata{
+		ID:      "",
+		Type:    "",
+		SubType: "",
+		Name:    "",
+	}
+}

resources/fetchers/file_system_fetcher.go

@@ -31,13 +31,20 @@ import (
 	"github.com/elastic/cloudbeat/resources/fetching"
 )
 
+const (
+	FSResourceType = "file"
+	FileSubType    = "file"
+	DirSubType     = "directory"
+)
+
 type FileSystemResource struct {
 	FileName string `json:"filename"`
 	FileMode string `json:"mode"`
 	Gid      string `json:"gid"`
 	Uid      string `json:"uid"`
 	Path     string `json:"path"`
 	Inode    string `json:"inode"`
+	SubType  string `json:"sub_type"`
 }
 
 // FileSystemFetcher implement the Fetcher interface
@@ -104,7 +111,7 @@ func FromFileInfo(info os.FileInfo, path string) (FileSystemResource, error) {
 	usr, _ := user.LookupId(u)
 	group, _ := user.LookupGroupId(g)
 	mod := strconv.FormatUint(uint64(info.Mode().Perm()), 8)
-	inode := strconv.FormatUint(uint64(stat.Ino), 10)
+	inode := strconv.FormatUint(stat.Ino, 10)
 
 	data := FileSystemResource{
 		FileName: info.Name(),
@@ -113,6 +120,7 @@ func FromFileInfo(info os.FileInfo, path string) (FileSystemResource, error) {
 		Gid:      group.Name,
 		Path:     path,
 		Inode:    inode,
+		SubType:  getSubType(info.IsDir()),
 	}
 
 	return data, nil
@@ -121,10 +129,22 @@ func FromFileInfo(info os.FileInfo, path string) (FileSystemResource, error) {
 func (f *FileSystemFetcher) Stop() {
 }
 
-func (r FileSystemResource) GetID() (string, error) {
-	return r.Inode, nil
-}
-
 func (r FileSystemResource) GetData() interface{} {
 	return r
 }
+
+func (r FileSystemResource) GetMetadata() fetching.ResourceMetadata {
+	return fetching.ResourceMetadata{
+		ID:      r.Inode,
+		Type:    FSResourceType,
+		SubType: r.SubType,
+		Name:    r.Path,
+	}
+}
+
+func getSubType(isDir bool) string {
+	if isDir {
+		return DirSubType
+	}
+	return FileSubType
+}

resources/fetchers/file_system_fetcher_test.go

@@ -48,9 +48,12 @@ func TestFileFetcherFetchASingleFile(t *testing.T) {
 	fsResource := results[0].(FileSystemResource)
 	assert.Equal(t, files[0], fsResource.FileName)
 	assert.Equal(t, "600", fsResource.FileMode)
-	rid, err := fsResource.GetID()
-	assert.NotNil(t, rid)
-	assert.NoError(t, err)
+
+	rMetadata := fsResource.GetMetadata()
+	assert.NotNil(t, rMetadata.ID)
+	assert.Equal(t, filePaths[0], rMetadata.Name)
+	assert.Equal(t, FileSubType, rMetadata.SubType)
+	assert.Equal(t, FSResourceType, rMetadata.Type)
 }
 
 func TestFileFetcherFetchTwoPatterns(t *testing.T) {
@@ -59,9 +62,9 @@ func TestFileFetcherFetchTwoPatterns(t *testing.T) {
 	outerDir := createDirectoriesWithFiles(t, "", outerDirectoryName, outerFiles)
 	defer os.RemoveAll(outerDir)
 
-	path := []string{filepath.Join(outerDir, outerFiles[0]), filepath.Join(outerDir, outerFiles[1])}
+	paths := []string{filepath.Join(outerDir, outerFiles[0]), filepath.Join(outerDir, outerFiles[1])}
 	cfg := FileFetcherConfig{
-		Patterns: path,
+		Patterns: paths,
 	}
 	factory := FileSystemFactory{}
 	fileFetcher, err := factory.CreateFrom(cfg)
@@ -74,16 +77,22 @@ func TestFileFetcherFetchTwoPatterns(t *testing.T) {
 	firstFSResource := results[0].(FileSystemResource)
 	assert.Equal(t, outerFiles[0], firstFSResource.FileName)
 	assert.Equal(t, "600", firstFSResource.FileMode)
-	rid, err := firstFSResource.GetID()
-	assert.NotNil(t, rid)
-	assert.NoError(t, err)
+
+	rMetadata := firstFSResource.GetMetadata()
+	assert.NotNil(t, rMetadata.ID)
+	assert.Equal(t, paths[0], rMetadata.Name)
+	assert.Equal(t, FileSubType, rMetadata.SubType)
+	assert.Equal(t, FSResourceType, rMetadata.Type)
 
 	secFSResource := results[1].(FileSystemResource)
 	assert.Equal(t, outerFiles[1], secFSResource.FileName)
 	assert.Equal(t, "600", secFSResource.FileMode)
-	rid, err = secFSResource.GetID()
-	assert.NotNil(t, rid)
-	assert.NoError(t, err)
+
+	SecResMetadata := secFSResource.GetMetadata()
+	assert.NotNil(t, SecResMetadata.ID)
+	assert.Equal(t, paths[1], SecResMetadata.Name)
+	assert.Equal(t, FileSubType, SecResMetadata.SubType)
+	assert.Equal(t, FSResourceType, SecResMetadata.Type)
 }
 
 func TestFileFetcherFetchDirectoryOnly(t *testing.T) {
@@ -105,12 +114,14 @@ func TestFileFetcherFetchDirectoryOnly(t *testing.T) {
 	assert.Equal(t, 1, len(results))
 
 	fsResource := results[0].(FileSystemResource)
-
 	expectedResult := filepath.Base(dir)
+	rMetadata := fsResource.GetMetadata()
+
 	assert.Equal(t, expectedResult, fsResource.FileName)
-	rid, err := fsResource.GetID()
-	assert.NotNil(t, rid)
-	assert.NoError(t, err)
+	assert.NotNil(t, rMetadata.ID)
+	assert.NotNil(t, rMetadata.Name)
+	assert.Equal(t, DirSubType, rMetadata.SubType)
+	assert.Equal(t, FSResourceType, rMetadata.Type)
 }
 
 func TestFileFetcherFetchOuterDirectoryOnly(t *testing.T) {
@@ -138,10 +149,13 @@ func TestFileFetcherFetchOuterDirectoryOnly(t *testing.T) {
 	//All inner files should exist in the final result
 	expectedResult := []string{"output.txt", filepath.Base(innerDir)}
 	for i := 0; i < len(results); i++ {
+		rMetadata := results[i].GetMetadata()
 		fileSystemDataResources := results[i].(FileSystemResource)
 		assert.Contains(t, expectedResult, fileSystemDataResources.FileName)
-		rid, err := results[i].GetID()
-		assert.NotNil(t, rid)
+		assert.NotNil(t, rMetadata.SubType)
+		assert.NotNil(t, rMetadata.Name)
+		assert.NotNil(t, rMetadata.ID)
+		assert.Equal(t, FSResourceType, rMetadata.Type)
 		assert.NoError(t, err)
 	}
 }
@@ -178,8 +192,11 @@ func TestFileFetcherFetchDirectoryRecursively(t *testing.T) {
 	//All inner files should exist in the final result
 	for i := 0; i < len(results); i++ {
 		fileSystemDataResources := results[i].(FileSystemResource)
-		rid, err := results[i].GetID()
-		assert.NotNil(t, rid)
+		rMetadata := results[i].GetMetadata()
+		assert.NotNil(t, rMetadata.SubType)
+		assert.NotNil(t, rMetadata.Name)
+		assert.NotNil(t, rMetadata.ID)
+		assert.Equal(t, FSResourceType, rMetadata.Type)
 		assert.NoError(t, err)
 		assert.Contains(t, allFilesName, fileSystemDataResources.FileName)
 	}

resources/fetchers/iam_fetcher.go

@@ -60,3 +60,13 @@ func (r IAMResource) GetID() (string, error) {
 func (r IAMResource) GetData() interface{} {
 	return r.Data
 }
+
+func (r IAMResource) GetMetadata() fetching.ResourceMetadata {
+	//TODO implement me
+	return fetching.ResourceMetadata{
+		ID:      "",
+		Type:    "",
+		SubType: "",
+		Name:    "",
+	}
+}

resources/fetchers/kube_provider.go

@@ -18,7 +18,6 @@
 package fetchers
 
 import (
-	"fmt"
 	"reflect"
 
 	"github.com/elastic/beats/v7/libbeat/common/kubernetes"
@@ -31,7 +30,11 @@ type K8sResource struct {
 	Data interface{}
 }
 
-const k8sObjMetadataField = "ObjectMeta"
+const (
+	k8sObjMetadataField  = "ObjectMeta"
+	k8sTypeMetadataField = "TypeMeta"
+	k8sObjType           = "k8s_object"
+)
 
 func GetKubeData(watchers []kubernetes.Watcher) []fetching.Resource {
 	logp.L().Info("Fetching Kubernetes data")
@@ -62,19 +65,43 @@ func GetKubeData(watchers []kubernetes.Watcher) []fetching.Resource {
 	return ret
 }
 
-func (r K8sResource) GetID() (string, error) {
+func (r K8sResource) GetData() interface{} {
+	return r.Data
+}
+
+func (r K8sResource) GetMetadata() fetching.ResourceMetadata {
+	k8sObjMeta := r.GetK8sObjectMeta()
+	resourceID := k8sObjMeta.UID
+	resourceName := k8sObjMeta.Name
+
+	return fetching.ResourceMetadata{
+		ID:      string(resourceID),
+		Type:    k8sObjType,
+		SubType: r.GetSubType(),
+		Name:    resourceName,
+	}
+}
+
+func (r K8sResource) GetK8sObjectMeta() metav1.ObjectMeta {
 	k8sObj := reflect.Indirect(reflect.ValueOf(r.Data))
 	metadata, ok := k8sObj.FieldByName(k8sObjMetadataField).Interface().(metav1.ObjectMeta)
 	if !ok {
-		return "", fmt.Errorf("failed to retrieve object metadata")
+		logp.L().Errorf("failed to retrieve object metadata, Resource: %#v", r)
+		return metav1.ObjectMeta{}
 	}
 
-	uid := metadata.UID
-	return string(uid), nil
+	return metadata
 }
 
-func (r K8sResource) GetData() interface{} {
-	return r.Data
+func (r K8sResource) GetSubType() string {
+	k8sObj := reflect.Indirect(reflect.ValueOf(r.Data))
+	typeMeta, ok := k8sObj.FieldByName(k8sTypeMetadataField).Interface().(metav1.TypeMeta)
+	if !ok {
+		logp.L().Errorf("failed to retrieve type metadata, Resource: %#v", r)
+		return ""
+	}
+
+	return typeMeta.Kind
 }
 
 // nullifyManagedFields ManagedFields field contains fields with dot that prevent from elasticsearch to index

resources/fetchers/process_fetcher.go

@@ -36,7 +36,9 @@ const (
 	// Expects format as the following: --<key><delimiter><value>.
 	// For example: --config=a.json
 	// The regex supports two delimiters "=" and ""
-	CMDArgumentMatcher = "\\b%s[\\s=]\\/?(\\S+)"
+	CMDArgumentMatcher  = "\\b%s[\\s=]\\/?(\\S+)"
+	ProcessResourceType = "process"
+	ProcessSubType      = "process"
 )
 
 type ProcessResource struct {
@@ -163,10 +165,15 @@ func (f *ProcessesFetcher) readConfigurationFile(path string, data []byte) (inte
 func (f *ProcessesFetcher) Stop() {
 }
 
-func (res ProcessResource) GetID() (string, error) {
-	return res.PID, nil
-}
-
 func (res ProcessResource) GetData() interface{} {
 	return res
 }
+
+func (res ProcessResource) GetMetadata() fetching.ResourceMetadata {
+	return fetching.ResourceMetadata{
+		ID:      res.PID,
+		Type:    ProcessResourceType,
+		SubType: ProcessSubType,
+		Name:    res.Stat.Name,
+	}
+}

resources/fetching/fetcher.go

@@ -40,10 +40,21 @@ type Condition interface {
 }
 
 type Resource interface {
-	GetID() (string, error)
+	GetMetadata() ResourceMetadata
 	GetData() interface{}
 }
 
+type ResourceFields struct {
+	ResourceMetadata
+	Raw interface{} `json:"raw"`
+}
+
+type ResourceMetadata struct {
+	ID      string `json:"id"`
+	Type    string `json:"type"`
+	SubType string `json:"sub_type"`
+	Name    string `json:"name"`
+}
 type Result struct {
 	Type string `json:"type"`
 	// Golang 1.18 will introduce generics which will be useful for typing the resource field