Merge pull request #222 from jetstack/informer-dynamic-datagathers

Introduces informers in the dynamic datagatherers to find the fetch resources on events rather than polling resources.

Author: Oluwole Fadeyi

api/datareading.go

@@ -1,6 +1,9 @@
 package api
 
-import "time"
+import (
+	"encoding/json"
+	"time"
+)
 
 // DataReadingsPost is the payload in the upload request.
 type DataReadingsPost struct {
@@ -14,8 +17,34 @@ type DataReadingsPost struct {
 type DataReading struct {
 	// ClusterID is optional as it can be infered from the agent
 	// token when using basic authentication.
-	ClusterID    string      `json:"cluster_id,omitempty"`
-	DataGatherer string      `json:"data-gatherer"`
-	Timestamp    Time        `json:"timestamp"`
-	Data         interface{} `json:"data"`
+	ClusterID     string      `json:"cluster_id,omitempty"`
+	DataGatherer  string      `json:"data-gatherer"`
+	Timestamp     Time        `json:"timestamp"`
+	Data          interface{} `json:"data"`
+	SchemaVersion string      `json:"schema_version"`
+}
+
+// GatheredResource wraps the raw k8s resource that is sent to the jetstack secure backend
+type GatheredResource struct {
+	// Resource is a reference to a k8s object that was found by the informer
+	// should be of type unstructured.Unstructured, raw Object
+	Resource  interface{}
+	DeletedAt Time
+}
+
+func (v GatheredResource) MarshalJSON() ([]byte, error) {
+	dateString := ""
+	if !v.DeletedAt.IsZero() {
+		dateString = v.DeletedAt.Format(TimeFormat)
+	}
+
+	data := struct {
+		Resource  interface{} `json:"resource"`
+		DeletedAt string      `json:"deleted_at,omitempty"`
+	}{
+		Resource:  v.Resource,
+		DeletedAt: dateString,
+	}
+
+	return json.Marshal(data)
 }

api/datareading_test.go

@@ -0,0 +1,36 @@
+package api
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+)
+
+func TestJSONGatheredResourceDropsEmptyTime(t *testing.T) {
+	var resource GatheredResource
+	bytes, err := json.Marshal(resource)
+	if err != nil {
+		t.Fatalf("failed to marshal %s", err)
+	}
+
+	expected := `{"resource":null}`
+
+	if string(bytes) != expected {
+		t.Fatalf("unexpected json \ngot  %s\nwant %s", string(bytes), expected)
+	}
+}
+
+func TestJSONGatheredResourceSetsTimeWhenPresent(t *testing.T) {
+	var resource GatheredResource
+	resource.DeletedAt = Time{time.Date(2021, 3, 29, 0, 0, 0, 0, time.UTC)}
+	bytes, err := json.Marshal(resource)
+	if err != nil {
+		t.Fatalf("failed to marshal %s", err)
+	}
+
+	expected := `{"resource":null,"deleted_at":"2021-03-29T00:00:00Z"}`
+
+	if string(bytes) != expected {
+		t.Fatalf("unexpected json \ngot  %s\nwant %s", string(bytes), expected)
+	}
+}

go.mod

@@ -21,6 +21,7 @@ require (
 	github.com/kylelemons/godebug v1.1.0
 	github.com/leodido/go-urn v1.2.0 // indirect
 	github.com/pkg/errors v0.9.1
+	github.com/pmylund/go-cache v2.1.0+incompatible
 	github.com/sirupsen/logrus v1.7.0
 	github.com/spf13/cobra v1.1.1
 	github.com/spf13/pflag v1.0.5

go.sum

@@ -779,6 +779,9 @@ github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE
 github.com/pkg/profile v1.2.1/go.mod h1:hJw3o1OdXxsrSjjVksARp5W95eeEaEfptyVZyv6JUPA=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/pmylund/go-cache v1.0.0 h1:jbJMNhn4LhBfb3dRejPlnjxSiokDt4qO5NWt8mMi+UE=
+github.com/pmylund/go-cache v2.1.0+incompatible h1:n+7K51jLz6a3sCvff3BppuCAkixuDHuJ/C57Vw/XjTE=
+github.com/pmylund/go-cache v2.1.0+incompatible/go.mod h1:hmz95dGvINpbRZGsqPcd7B5xXY5+EKb5PpGhQY3NTHk=
 github.com/posener/complete v1.1.1/go.mod h1:em0nMJCgc9GFtwrmVmEMR/ZL6WyhyjMBndrE9hABlRI=
 github.com/pquerna/cachecontrol v0.0.0-20171018203845-0dec1b30a021 h1:0XM1XL/OFFJjXsYXlG30spTkV/E9+gmd5GD1w2HE8xM=
 github.com/pquerna/cachecontrol v0.0.0-20171018203845-0dec1b30a021/go.mod h1:prYjPmNq4d1NPVmpShWobRqXY3q7Vp+80DqgxxUrUIA=

pkg/agent/dummy_data_gatherer.go

@@ -29,6 +29,21 @@ type dummyDataGatherer struct {
 	FailedAttempts int
 }
 
+func (g *dummyDataGatherer) Run(stopCh <-chan struct{}) error {
+	// no async functionality, see Fetch
+	return nil
+}
+
+func (g *dummyDataGatherer) WaitForCacheSync(stopCh <-chan struct{}) error {
+	// no async functionality, see Fetch
+	return nil
+}
+
+func (g *dummyDataGatherer) Delete() error {
+	// no async functionality, see Fetch
+	return nil
+}
+
 func (c *dummyDataGatherer) Fetch() (interface{}, error) {
 	var err error
 	if c.attemptNumber < c.FailedAttempts {