Use s3 caching instaed of elasticsearch (#45)

* use s3 caching instaed of elasticsearch

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

* clean up

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

Co-authored-by: Ayman <[email protected]>

Author: khalifapro

cmd/git/git.go

@@ -18,6 +18,7 @@ import (
 
 	"github.com/LF-Engineering/insights-datasource-git/build"
 	shared "github.com/LF-Engineering/insights-datasource-shared"
+	"github.com/LF-Engineering/insights-datasource-shared/cache"
 	elastic "github.com/LF-Engineering/insights-datasource-shared/elastic"
 	logger "github.com/LF-Engineering/insights-datasource-shared/ingestjob"
 	"github.com/LF-Engineering/lfx-event-schema/service"
@@ -579,6 +580,8 @@ type DSGit struct {
 	// RepositorySource for example git, github or gerrit
 	RepositorySource string
 	log              *logrus.Entry
+	cacheProvider    cache.Manager
+	endpoint         string
 }
 
 // PublisherPushEvents - this is a fake function to test publisher locally
@@ -1769,7 +1772,10 @@ func (j *DSGit) GitEnrichItems(ctx *shared.Ctx, thrN int, items []interface{}, d
 			*docs = []interface{}{}
 			gMaxUpstreamDtMtx.Lock()
 			defer gMaxUpstreamDtMtx.Unlock()
-			shared.SetLastUpdate(ctx, j.URL, gMaxUpstreamDt)
+			err = j.cacheProvider.SetLastSync(j.endpoint, gMaxUpstreamDt)
+			if err != nil {
+				return
+			}
 		}
 	}
 	if final {
@@ -2328,7 +2334,12 @@ func (j *DSGit) Sync(ctx *shared.Ctx) (err error) {
 		j.log.WithFields(logrus.Fields{"operation": "Sync"}).Infof("%s fetching from %v (%d threads)", j.URL, ctx.DateFrom, thrN)
 	}
 	if ctx.DateFrom == nil {
-		ctx.DateFrom = shared.GetLastUpdate(ctx, j.URL)
+		cachedLastSync, er := j.cacheProvider.GetLastSync(j.endpoint)
+		if er != nil {
+			err = er
+			return
+		}
+		ctx.DateFrom = &cachedLastSync
 		if ctx.DateFrom != nil {
 			j.log.WithFields(logrus.Fields{"operation": "Sync"}).Infof("%s resuming from %v (%d threads)", j.URL, ctx.DateFrom, thrN)
 		}
@@ -2591,7 +2602,7 @@ func (j *DSGit) Sync(ctx *shared.Ctx) (err error) {
 	// NOTE: Non-generic ends here
 	gMaxUpstreamDtMtx.Lock()
 	defer gMaxUpstreamDtMtx.Unlock()
-	shared.SetLastUpdate(ctx, j.URL, gMaxUpstreamDt)
+	err = j.cacheProvider.SetLastSync(j.endpoint, gMaxUpstreamDt)
 	return
 }
 
@@ -2600,7 +2611,7 @@ func main() {
 		ctx shared.Ctx
 		git DSGit
 	)
-	git.initStructuredLogger()
+	git.initStructuredLogger(&ctx)
 	err := git.Init(&ctx)
 	if err != nil {
 		git.log.WithFields(logrus.Fields{"operation": "main"}).Errorf("Error: %+v", err)
@@ -2610,6 +2621,7 @@ func main() {
 	shared.SetSyncMode(true, false)
 	shared.SetLogLoggerError(false)
 	shared.AddLogger(&git.Logger, GitDataSource, logger.Internal, []map[string]string{{"REPO_URL": git.URL, "ProjectSlug": ctx.Project}})
+	git.AddCacheProvider()
 	git.WriteLog(&ctx, timestamp, logger.InProgress, "")
 	err = git.Sync(&ctx)
 	if err != nil {
@@ -2621,15 +2633,29 @@ func main() {
 }
 
 // createStructuredLogger...
-func (j *DSGit) initStructuredLogger() {
+func (j *DSGit) initStructuredLogger(ctx *shared.Ctx) {
+	endpointURL := ""
+	if shared.FlagPassed(ctx, "url") && *j.FlagURL != "" {
+		endpointURL = *j.FlagURL
+	}
+	if ctx.EnvSet("URL") {
+		endpointURL = ctx.Env("URL")
+	}
 	logrus.SetFormatter(&logrus.JSONFormatter{})
 	log := logrus.WithFields(
 		logrus.Fields{
 			"environment": os.Getenv("STAGE"),
 			"commit":      build.GitCommit,
 			"version":     build.Version,
 			"service":     build.AppName,
-			"endpoint":    j.URL,
+			"endpoint":    endpointURL,
 		})
 	j.log = log
 }
+
+// AddCacheProvider - adds cache provider
+func (j *DSGit) AddCacheProvider() {
+	cacheProvider := cache.NewManager(GitDataSource, os.Getenv("STAGE"))
+	j.cacheProvider = *cacheProvider
+	j.endpoint = strings.ReplaceAll(strings.TrimPrefix(strings.TrimPrefix(j.URL, "https://"), "http://"), "/", "-")
+}

go.mod

@@ -3,7 +3,7 @@ module github.com/LF-Engineering/insights-datasource-git
 go 1.15
 
 require (
-	github.com/LF-Engineering/insights-datasource-shared v1.4.5-0.20220511063206-6754a12066b9
+	github.com/LF-Engineering/insights-datasource-shared v1.5.5
 	github.com/LF-Engineering/lfx-event-schema v0.1.20-0.20220511083836-c30a2d7cf561
 	github.com/aws/aws-lambda-go v1.27.1
 	github.com/aws/aws-sdk-go v1.42.25

go.sum

@@ -1,6 +1,6 @@
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
-github.com/LF-Engineering/insights-datasource-shared v1.4.5-0.20220511063206-6754a12066b9 h1:CcKhPF0SgXbOFasPhoD/1of1ohvgLc4aASJj2KTdsrI=
-github.com/LF-Engineering/insights-datasource-shared v1.4.5-0.20220511063206-6754a12066b9/go.mod h1:ZQCRAJNyizhBemQDSjqm14G2MPkOJbRzCPfjwFhYKy4=
+github.com/LF-Engineering/insights-datasource-shared v1.5.5 h1:jlSldY2MTR4gy82uQTed1M738gWrZuJj+EhAcYBrm3M=
+github.com/LF-Engineering/insights-datasource-shared v1.5.5/go.mod h1:ZQCRAJNyizhBemQDSjqm14G2MPkOJbRzCPfjwFhYKy4=
 github.com/LF-Engineering/lfx-event-schema v0.1.14/go.mod h1:CfFIZ4mwzo88umf5+KxDQEzqlVkPG7Vx8eLK2oDfWIs=
 github.com/LF-Engineering/lfx-event-schema v0.1.20-0.20220511083836-c30a2d7cf561 h1:6UCZsa14yxMNPcUN2S8Pgnb7WwOfs+N3UugFuW1tJAg=
 github.com/LF-Engineering/lfx-event-schema v0.1.20-0.20220511083836-c30a2d7cf561/go.mod h1:CfFIZ4mwzo88umf5+KxDQEzqlVkPG7Vx8eLK2oDfWIs=
@@ -80,6 +80,7 @@ github.com/urfave/cli/v2 v2.2.0/go.mod h1:SE9GqnLQmjVa0iPEY0f1w3ygNIYcIJ0OKPMoW2
 github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
 github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415/go.mod h1:GwrjFmJcFw6At/Gs6z4yjiIwzuJ1/+UwLxMQDVQXShQ=
 github.com/xeipuuv/gojsonschema v1.2.0/go.mod h1:anYRn/JVcOK2ZgGU+IjEV4nwlhoK5sQluxsYJ78Id3Y=
+golang.org/x/crypto v0.0.0-20211215153901-e495a2d5b3d3 h1:0es+/5331RGQPcXlMfP+WrnIIS6dNnNRe0WB02W0F4M=
 golang.org/x/crypto v0.0.0-20211215153901-e495a2d5b3d3/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/net v0.0.0-20181220203305-927f97764cc3/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=