RHCLOUD-35843 | refactor: Sources' REST API client

config/config.go

@@ -11,21 +11,22 @@ import (
 
 // SuperKeyWorkerConfig is the struct for storing runtime configuration
 type SuperKeyWorkerConfig struct {
-	Hostname           string
-	KafkaBrokerConfig  []clowder.BrokerConfig
-	KafkaTopics        map[string]string
-	KafkaGroupID       string
-	MetricsPort        int
-	LogLevel           string
-	LogGroup           string
-	LogHandler         string
-	AwsRegion          string
-	AwsAccessKeyID     string
-	AwsSecretAccessKey string
-	SourcesHost        string
-	SourcesScheme      string
-	SourcesPort        int
-	SourcesPSK         string
+	Hostname                   string
+	KafkaBrokerConfig          []clowder.BrokerConfig
+	KafkaTopics                map[string]string
+	KafkaGroupID               string
+	MetricsPort                int
+	LogLevel                   string
+	LogGroup                   string
+	LogHandler                 string
+	AwsRegion                  string
+	AwsAccessKeyID             string
+	AwsSecretAccessKey         string
+	SourcesHost                string
+	SourcesScheme              string
+	SourcesPort                int
+	SourcesPSK                 string
+	SourcesRequestsMaxAttempts int
 }
 
 // Get - returns the config parsed from runtime vars
@@ -81,6 +82,20 @@ func Get() *SuperKeyWorkerConfig {
 	options.SetDefault("SourcesPort", os.Getenv("SOURCES_PORT"))
 	options.SetDefault("SourcesPSK", os.Getenv("SOURCES_PSK"))
 
+	// Get the number of maximum request attempts we want to make to the Sources' API.
+	sourcesRequestsMaxAttempts, err := strconv.Atoi(os.Getenv("SOURCES_REQUEST_MAX_ATTEMPTS"))
+	if err != nil {
+		log.Printf(`Warning: the provided max attempts value \"%s\" is not an integer. Setting default value of 3.`, os.Getenv("SOURCES_REQUEST_MAX_ATTEMPTS"))
+		sourcesRequestsMaxAttempts = 3
+	}
+
+	if sourcesRequestsMaxAttempts < 1 {
+		log.Printf(`Warning: the provided max attempts value \"%s\" is lower than 1, and we need to at least make one attempt when calling Sources. Setting default value of 3.`, os.Getenv("SOURCES_REQUEST_MAX_ATTEMPTS"))
+		sourcesRequestsMaxAttempts = 3
+	}
+
+	options.SetDefault("SourcesRequestsMaxAttempts", sourcesRequestsMaxAttempts)
+
 	hostname, _ := os.Hostname()
 	options.SetDefault("Hostname", hostname)
 
@@ -94,21 +109,22 @@ func Get() *SuperKeyWorkerConfig {
 	options.AutomaticEnv()
 
 	return &SuperKeyWorkerConfig{
-		Hostname:           options.GetString("Hostname"),
-		KafkaBrokerConfig:  brokerConfig,
-		KafkaTopics:        options.GetStringMapString("KafkaTopics"),
-		KafkaGroupID:       options.GetString("KafkaGroupID"),
-		MetricsPort:        options.GetInt("MetricsPort"),
-		LogLevel:           options.GetString("LogLevel"),
-		LogHandler:         options.GetString("LogHandler"),
-		LogGroup:           options.GetString("LogGroup"),
-		AwsRegion:          options.GetString("AwsRegion"),
-		AwsAccessKeyID:     options.GetString("AwsAccessKeyID"),
-		AwsSecretAccessKey: options.GetString("AwsSecretAccessKey"),
-		SourcesHost:        options.GetString("SourcesHost"),
-		SourcesScheme:      options.GetString("SourcesScheme"),
-		SourcesPort:        options.GetInt("SourcesPort"),
-		SourcesPSK:         options.GetString("SourcesPSK"),
+		Hostname:                   options.GetString("Hostname"),
+		KafkaBrokerConfig:          brokerConfig,
+		KafkaTopics:                options.GetStringMapString("KafkaTopics"),
+		KafkaGroupID:               options.GetString("KafkaGroupID"),
+		MetricsPort:                options.GetInt("MetricsPort"),
+		LogLevel:                   options.GetString("LogLevel"),
+		LogHandler:                 options.GetString("LogHandler"),
+		LogGroup:                   options.GetString("LogGroup"),
+		AwsRegion:                  options.GetString("AwsRegion"),
+		AwsAccessKeyID:             options.GetString("AwsAccessKeyID"),
+		AwsSecretAccessKey:         options.GetString("AwsSecretAccessKey"),
+		SourcesHost:                options.GetString("SourcesHost"),
+		SourcesScheme:              options.GetString("SourcesScheme"),
+		SourcesPort:                options.GetInt("SourcesPort"),
+		SourcesPSK:                 options.GetString("SourcesPSK"),
+		SourcesRequestsMaxAttempts: options.GetInt("SourcesRequestsMaxAttempts"),
 	}
 }
 

deploy/clowdapp.yaml

@@ -114,3 +114,6 @@ parameters:
   displayName: AWS Wait Time
   description: Time to sleep between creating the resources and posting back to Sources API
   value: "15"
+- name: SOURCES_REQUEST_MAX_ATTEMPTS
+  description: The maximum request attempts to make when calling the Sources API.
+  value: "3"

logger/logger_context.go

@@ -2,6 +2,8 @@ package logger
 
 import (
 	"context"
+	"net/http"
+	"net/url"
 
 	"github.com/sirupsen/logrus"
 )
@@ -22,6 +24,30 @@ type applicationIdCtxKeyType string
 // or fetching the variable to/from the context.
 type applicationTypeCtxKeyType string
 
+// authenticationIdCtxKeyType defines the type for the authentication's identifier key that will ensure type safety
+// when storing or fetching the variable to/from the context.
+type authenticationIdCtxKeyType string
+
+// resourceTypeCtxKeyType defines the type for the resource's type key that will ensure type safety when storing or
+// fetching the variable to/from the context.
+type resourceTypeCtxKeyType string
+
+// resourceIdCtxKeyType defines the type for the resource's identifier key that will ensure type safety when storing or
+// fetching the variable to/from the context.
+type resourceIdCtxKeyType string
+
+// httpMethodCtxKeyType defines the type for the HTTP method key that will ensure type safety when storing or
+// fetching the variable to/from the context.
+type httpMethodCtxKeyType string
+
+// urlCtxKeyType defines the type for the URL key that will ensure type safety when storing or fetching the variable
+// to/from the context.
+type urlCtxKeyType string
+
+// urlCtxKeyType defines the type for the HTTP headers key that will ensure type safety when storing or fetching the
+// variable to/from the context.
+type httpHeadersCtxKeyType string
+
 // tenantIdCtxKey defines the key to be used to store the tenant's identifier.
 const tenantIdCtxKey tenantIdCtxKeyType = "tenant_id"
 
@@ -34,6 +60,24 @@ const applicationIdCtxKey applicationIdCtxKeyType = "application_id"
 // applicationTypeCtxKey defines the key to be used to store the application type's identifier.
 const applicationTypeCtxKey applicationTypeCtxKeyType = "application_type"
 
+// authenticationIdCtxKey defines the key to be used to store the authentication's identifier.
+const authenticationIdCtxKey authenticationIdCtxKeyType = "authentication_id"
+
+// resourceTypeCtxKey defines the key to be used to store the resource's type.
+const resourceTypeCtxKey resourceTypeCtxKeyType = "resource_type"
+
+// resourceIdCtxKey defines the key to be used to store the resource's identifier.
+const resourceIdCtxKey resourceIdCtxKeyType = "resource_id"
+
+// httpMethodCtxKey defines the key to be used to store the HTTP method's identifier.
+const httpMethodCtxKey httpMethodCtxKeyType = "http_method"
+
+// urlCtxKey defines the key to be used to store the URL's identifier.
+const urlCtxKey urlCtxKeyType = "url"
+
+// httpHeadersCtxKey defines the key to be used to store the HTTP headers' identifier.
+const httpHeadersCtxKey httpHeadersCtxKeyType = "http_headers"
+
 // LogWithContext returns a logger with all the fields defined in the context.
 func LogWithContext(ctx context.Context) *logrus.Entry {
 	logFields := logrus.Fields{}
@@ -54,6 +98,30 @@ func LogWithContext(ctx context.Context) *logrus.Entry {
 		logFields["application_type"] = applicationType
 	}
 
+	if authenticationId, ok := ctx.Value(authenticationIdCtxKey).(authenticationIdCtxKeyType); ok {
+		logFields["authentication_id"] = authenticationId
+	}
+
+	if resourceType, ok := ctx.Value(resourceTypeCtxKey).(resourceTypeCtxKeyType); ok {
+		logFields["resource_type"] = resourceType
+	}
+
+	if resourceId, ok := ctx.Value(resourceIdCtxKey).(resourceIdCtxKeyType); ok {
+		logFields["resource_id"] = resourceId
+	}
+
+	if httpMethod, ok := ctx.Value(httpMethodCtxKey).(authenticationIdCtxKeyType); ok {
+		logFields["http_method"] = httpMethod
+	}
+
+	if urlVar, ok := ctx.Value(urlCtxKey).(urlCtxKeyType); ok {
+		logFields["url"] = urlVar
+	}
+
+	if httpHeaders, ok := ctx.Value(httpHeadersCtxKey).(httpHeadersCtxKeyType); ok {
+		logFields["http_headers"] = httpHeaders
+	}
+
 	return Log.WithFields(logFields)
 }
 
@@ -77,3 +145,34 @@ func WithApplicationId(ctx context.Context, applicationId string) context.Contex
 func WithApplicationType(ctx context.Context, applicationType string) context.Context {
 	return context.WithValue(ctx, applicationTypeCtxKey, applicationType)
 }
+
+// WithAuthenticationId creates a new context by copying the given context and appending the authentication's
+// identifier to it.
+func WithAuthenticationId(ctx context.Context, authenticationId string) context.Context {
+	return context.WithValue(ctx, authenticationIdCtxKey, authenticationId)
+}
+
+// WithResourceType creates a new context by copying the given context and appending the resource's type to it.
+func WithResourceType(ctx context.Context, resourceType string) context.Context {
+	return context.WithValue(ctx, resourceTypeCtxKey, resourceType)
+}
+
+// WithResourceId creates a new context by copying the given context and appending the resource's identifier to it.
+func WithResourceId(ctx context.Context, resourceId string) context.Context {
+	return context.WithValue(ctx, resourceIdCtxKey, resourceId)
+}
+
+// WithHttpMethod creates a new context by copying the given context and appending the HTTP method to it.
+func WithHttpMethod(ctx context.Context, httpMethod string) context.Context {
+	return context.WithValue(ctx, httpMethodCtxKey, httpMethod)
+}
+
+// WithURL creates a new context by copying the given context and appending the URL to it.
+func WithURL(ctx context.Context, url *url.URL) context.Context {
+	return context.WithValue(ctx, urlCtxKey, url.String())
+}
+
+// WithHTTPHeaders creates a new context by copying the given context and appending the HTTP headers to it.
+func WithHTTPHeaders(ctx context.Context, headers http.Header) context.Context {
+	return context.WithValue(ctx, httpHeadersCtxKey, headers)
+}

provider/forge.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 
 	"github.com/redhatinsights/sources-superkey-worker/amazon"
+	"github.com/redhatinsights/sources-superkey-worker/config"
 	"github.com/redhatinsights/sources-superkey-worker/sources"
 	"github.com/redhatinsights/sources-superkey-worker/superkey"
 )
@@ -45,8 +46,14 @@ func TearDown(ctx context.Context, f *superkey.ForgedApplication) []error {
 
 // getProvider returns a provider based on create request's provider + credentials
 func getProvider(ctx context.Context, request *superkey.CreateRequest) (superkey.Provider, error) {
-	client := sources.SourcesClient{AccountNumber: request.TenantID, IdentityHeader: request.IdentityHeader, OrgId: request.OrgIdHeader}
-	auth, err := client.GetInternalAuthentication(ctx, request.SuperKey)
+	sourcesRestClient := sources.NewSourcesClient(config.Get())
+
+	authData := sources.AuthenticationData{
+		IdentityHeader: request.IdentityHeader,
+		OrgId:          request.OrgIdHeader,
+	}
+
+	auth, err := sourcesRestClient.GetInternalAuthentication(ctx, &authData, request.SuperKey)
 	if err != nil {
 		return nil, fmt.Errorf(`error while fetching internal authentication "%s" from Sources: %w`, request.SuperKey, err)
 	}