PUT and POST APIs for tenant creation

internal/pkg/app/app.go

@@ -253,44 +253,48 @@ func SetupAPIServer(lifecycle fx.Lifecycle, config config.Config, logger *zerolo
 		Handler: mux,
 	}
 
-	// initialize google profiler
-	profilerConfigMap := make(map[string]string, 0)
-	profilerConfigMap["type"] = config.GetString("ears.profiler.type")
-	profilerConfigMap["project_id"] = config.GetString("ears.profiler.project_id")
-	profilerConfigMap["private_key_id"] = config.GetString("ears.profiler.private_key_id")
-	profilerConfigMap["private_key"] = config.GetString("ears.profiler.private_key")
-	profilerConfigMap["client_email"] = config.GetString("ears.profiler.client_email")
-	profilerConfigMap["client_id"] = config.GetString("ears.profiler.client_id")
-	profilerConfigMap["auth_uri"] = config.GetString("ears.profiler.auth_uri")
-	profilerConfigMap["token_uri"] = config.GetString("ears.profiler.token_uri")
-	profilerConfigMap["auth_provider_x509_cert_url"] = config.GetString("ears.profiler.auth_provider_x509_cert_url")
-	profilerConfigMap["client_x509_cert_url"] = config.GetString("ears.profiler.client_x509_cert_url")
-	profilerConfigMap["universe_domain"] = config.GetString("ears.profiler.universe_domain")
-	buf, err := json.MarshalIndent(profilerConfigMap, "", "\t")
-	if err != nil {
-		logger.Error().Str("op", "SetupAPIServer.StartProfiler").Msg(err.Error())
-	}
-	err = os.WriteFile("google_profiler_config.json", buf, 0644)
-	if err != nil {
-		logger.Error().Str("op", "SetupAPIServer.StartProfiler").Msg(err.Error())
-	}
-	workingDir, err := os.Getwd()
-	if err != nil {
-		logger.Error().Str("op", "SetupAPIServer.StartProfiler").Msg(err.Error())
-	}
-	os.Setenv("GOOGLE_APPLICATION_CREDENTIALS", filepath.Join(workingDir, "google_profiler_config.json"))
-	taskArn, err := GetTaskArn()
-	if err != nil {
-		logger.Error().Str("op", "SetupAPIServer.StartProfiler").Msg(err.Error())
-	}
-	cfg := profiler.Config{
-		Service:        "ears",
-		ServiceVersion: app.Version,
-		ProjectID:      "ears-project",
-		Zone:           config.GetString("ears.env") + "-" + taskArn,
-	}
-	if err := profiler.Start(cfg); err != nil {
-		logger.Error().Str("op", "SetupAPIServer.StartProfiler").Msg(err.Error())
+	if config.GetString("ears.profiler.active") == "yes" {
+		// initialize google profiler
+		profilerConfigMap := make(map[string]string, 0)
+		profilerConfigMap["type"] = config.GetString("ears.profiler.type")
+		profilerConfigMap["project_id"] = config.GetString("ears.profiler.project_id")
+		profilerConfigMap["private_key_id"] = config.GetString("ears.profiler.private_key_id")
+		profilerConfigMap["private_key"] = config.GetString("ears.profiler.private_key")
+		profilerConfigMap["client_email"] = config.GetString("ears.profiler.client_email")
+		profilerConfigMap["client_id"] = config.GetString("ears.profiler.client_id")
+		profilerConfigMap["auth_uri"] = config.GetString("ears.profiler.auth_uri")
+		profilerConfigMap["token_uri"] = config.GetString("ears.profiler.token_uri")
+		profilerConfigMap["auth_provider_x509_cert_url"] = config.GetString("ears.profiler.auth_provider_x509_cert_url")
+		profilerConfigMap["client_x509_cert_url"] = config.GetString("ears.profiler.client_x509_cert_url")
+		profilerConfigMap["universe_domain"] = config.GetString("ears.profiler.universe_domain")
+		buf, err := json.MarshalIndent(profilerConfigMap, "", "\t")
+		if err != nil {
+			logger.Error().Str("op", "SetupAPIServer.StartProfiler").Msg(err.Error())
+		}
+		err = os.WriteFile("google_profiler_config.json", buf, 0644)
+		if err != nil {
+			logger.Error().Str("op", "SetupAPIServer.StartProfiler").Msg(err.Error())
+		}
+		workingDir, err := os.Getwd()
+		if err != nil {
+			logger.Error().Str("op", "SetupAPIServer.StartProfiler").Msg(err.Error())
+		}
+		os.Setenv("GOOGLE_APPLICATION_CREDENTIALS", filepath.Join(workingDir, "google_profiler_config.json"))
+		taskArn, err := GetTaskArn()
+		if err != nil {
+			logger.Error().Str("op", "SetupAPIServer.StartProfiler").Msg(err.Error())
+		}
+		cfg := profiler.Config{
+			Service:        "ears",
+			ServiceVersion: app.Version,
+			ProjectID:      "ears-project",
+			Zone:           config.GetString("ears.env") + "-" + taskArn,
+		}
+		if err := profiler.Start(cfg); err != nil {
+			logger.Error().Str("op", "SetupAPIServer.StartProfiler").Msg(err.Error())
+		}
+	} else {
+		logger.Info().Msg("Google profiler disabled")
 	}
 
 	// initialize event logger

internal/pkg/app/handlers_v1.go

@@ -36,7 +36,7 @@ import (
 	"go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/metric/global"
 	"go.opentelemetry.io/otel/trace"
-	"io/ioutil"
+	"io"
 	"net/http"
 	"regexp"
 	"strings"
@@ -175,12 +175,16 @@ func NewAPIManager(routingMgr tablemgr.RoutingTableManager, tenantStorer tenant.
 	api.muxRouter.HandleFunc("/ears/v1/orgs/{orgId}/applications/{appId}/fragments/{fragmentId}", api.getFragmentHandler).Methods(http.MethodGet)
 	api.muxRouter.HandleFunc("/ears/v1/orgs/{orgId}/applications/{appId}/fragments", api.getAllTenantFragmentsHandler).Methods(http.MethodGet)
 
+	// old tenant APIs for backward compatibility
+
 	api.muxRouter.HandleFunc("/ears/v1/orgs/{orgId}/applications/{appId}/config", api.getTenantConfigHandler).Methods(http.MethodGet)
-	api.muxRouter.HandleFunc("/ears/v1/orgs/{orgId}/applications/{appId}/config", api.addTenantConfigHandler).Methods(http.MethodPut)
+	api.muxRouter.HandleFunc("/ears/v1/orgs/{orgId}/applications/{appId}/config", api.updateTenantConfigHandler).Methods(http.MethodPut)
+	api.muxRouter.HandleFunc("/ears/v1/orgs/{orgId}/applications/{appId}/config", api.addTenantConfigHandler).Methods(http.MethodPost)
 	api.muxRouter.HandleFunc("/ears/v1/orgs/{orgId}/applications/{appId}/config", api.deleteTenantConfigHandler).Methods(http.MethodDelete)
 
 	api.muxRouter.HandleFunc("/ears/v1/orgs/{orgId}/applications/{appId}", api.getTenantConfigHandler).Methods(http.MethodGet)
-	api.muxRouter.HandleFunc("/ears/v1/orgs/{orgId}/applications/{appId}", api.addTenantConfigHandler).Methods(http.MethodPut)
+	api.muxRouter.HandleFunc("/ears/v1/orgs/{orgId}/applications/{appId}", api.updateTenantConfigHandler).Methods(http.MethodPut)
+	api.muxRouter.HandleFunc("/ears/v1/orgs/{orgId}/applications/{appId}", api.addTenantConfigHandler).Methods(http.MethodPost)
 	api.muxRouter.HandleFunc("/ears/v1/orgs/{orgId}/applications/{appId}", api.deleteTenantConfigHandler).Methods(http.MethodDelete)
 
 	api.muxRouter.HandleFunc("/ears/v1/routes", api.getAllRoutesHandler).Methods(http.MethodGet)
@@ -327,7 +331,7 @@ func (a *APIManager) webhookHandler(w http.ResponseWriter, r *http.Request) {
 		resp.Respond(ctx, w, doYaml(r))
 		return
 	}
-	body, err := ioutil.ReadAll(r.Body)
+	body, err := io.ReadAll(r.Body)
 	if err != nil {
 		log.Ctx(ctx).Error().Str("op", "webhookHandler").Msg(err.Error())
 		resp := ErrorResponse(&InternalServerError{err})
@@ -387,7 +391,7 @@ func (a *APIManager) sendEventHandler(w http.ResponseWriter, r *http.Request) {
 		r.Body = http.MaxBytesReader(w, r.Body, maxEventSize)
 		defer r.Body.Close()
 	}
-	body, err := ioutil.ReadAll(r.Body)
+	body, err := io.ReadAll(r.Body)
 	if err != nil {
 		log.Ctx(ctx).Error().Str("op", "sendEventHandler").Msg(err.Error())
 		resp := ErrorResponse(&InternalServerError{err})
@@ -506,7 +510,7 @@ func (a *APIManager) addRouteHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	routeId := vars["routeId"]
-	body, err := ioutil.ReadAll(r.Body)
+	body, err := io.ReadAll(r.Body)
 	if err != nil {
 		log.Ctx(ctx).Error().Str("op", "addRouteHandler").Msg(err.Error())
 		a.addRouteFailureRecorder.Add(ctx, 1.0)
@@ -872,7 +876,7 @@ func (a *APIManager) addFragmentHandler(w http.ResponseWriter, r *http.Request)
 		return
 	}
 	fragmentId := vars["fragmentId"]
-	body, err := ioutil.ReadAll(r.Body)
+	body, err := io.ReadAll(r.Body)
 	if err != nil {
 		log.Ctx(ctx).Error().Str("op", "addFragmentHandler").Msg(err.Error())
 		a.addRouteFailureRecorder.Add(ctx, 1.0)
@@ -964,7 +968,15 @@ func (a *APIManager) addTenantConfigHandler(w http.ResponseWriter, r *http.Reque
 		resp.Respond(ctx, w, doYaml(r))
 		return
 	}
-	body, err := ioutil.ReadAll(r.Body)
+	config, err := a.tenantStorer.GetConfig(ctx, *tid)
+	if err == nil && config != nil {
+		err = errors.New("cannot update existing tenant " + tid.AppId)
+		log.Ctx(ctx).Error().Str("op", "addTenantConfigHandler").Str("error", err.Error()).Msg("tenant " + tid.AppId + "already exists")
+		resp := ErrorResponse(convertToApiError(ctx, err))
+		resp.Respond(ctx, w, doYaml(r))
+		return
+	}
+	body, err := io.ReadAll(r.Body)
 	if err != nil {
 		log.Ctx(ctx).Error().Str("op", "addTenantConfigHandler").Str("error", err.Error()).Msg("error reading request body")
 		resp := ErrorResponse(&InternalServerError{err})
@@ -993,6 +1005,45 @@ func (a *APIManager) addTenantConfigHandler(w http.ResponseWriter, r *http.Reque
 	resp.Respond(ctx, w, doYaml(r))
 }
 
+func (a *APIManager) updateTenantConfigHandler(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+	vars := mux.Vars(r)
+	tid, apiErr := getTenant(ctx, vars)
+	if apiErr != nil {
+		log.Ctx(ctx).Error().Str("op", "updateTenantConfigHandler").Str("error", apiErr.Error()).Msg("orgId or appId empty")
+		resp := ErrorResponse(apiErr)
+		resp.Respond(ctx, w, doYaml(r))
+		return
+	}
+	body, err := io.ReadAll(r.Body)
+	if err != nil {
+		log.Ctx(ctx).Error().Str("op", "updateTenantConfigHandler").Str("error", err.Error()).Msg("error reading request body")
+		resp := ErrorResponse(&InternalServerError{err})
+		resp.Respond(ctx, w, doYaml(r))
+		return
+	}
+	var tenantConfig tenant.Config
+	err = yaml.Unmarshal(body, &tenantConfig)
+	if err != nil {
+		log.Ctx(ctx).Error().Str("op", "updateTenantConfigHandler").Str("error", err.Error()).Msg("error unmarshal request body")
+		resp := ErrorResponse(&BadRequestError{"Cannot unmarshal request body", err})
+		resp.Respond(ctx, w, doYaml(r))
+		return
+	}
+	tenantConfig.Tenant = *tid
+	err = a.tenantStorer.SetConfig(ctx, tenantConfig)
+	if err != nil {
+		log.Ctx(ctx).Error().Str("op", "updateTenantConfigHandler").Str("error", err.Error()).Msg("error setting tenant config")
+		resp := ErrorResponse(convertToApiError(ctx, err))
+		resp.Respond(ctx, w, doYaml(r))
+		return
+	}
+	a.quotaManager.PublishQuota(ctx, *tid)
+	log.Ctx(ctx).Info().Str("op", "updateTenantConfigHandler").Msg("success")
+	resp := ItemResponse(tenantConfig)
+	resp.Respond(ctx, w, doYaml(r))
+}
+
 func (a *APIManager) deleteTenantConfigHandler(w http.ResponseWriter, r *http.Request) {
 	ctx := r.Context()
 	vars := mux.Vars(r)

internal/pkg/app/middlewares.go

@@ -38,7 +38,10 @@ var eventUrlValidator = regexp.MustCompile(`^\/ears\/v1\/orgs\/[a-zA-Z0-9][a-zA-
 func NewMiddleware(logger *zerolog.Logger, jwtManager jwt.JWTConsumer) []func(next http.Handler) http.Handler {
 	middlewareLogger = logger
 	jwtMgr = jwtManager
+	// for this extraction to work, X-B3-TraceId header must be present and exactly 32 byte hex and X-B3-SpanId
+	// must be present and exactly 16 byte hex
 	otelMiddleware := otelmux.Middleware("ears", otelmux.WithPropagators(b3.New()))
+	//otelMiddleware := otelmux.Middleware("ears", otelmux.WithPropagators(otel.GetTextMapPropagator()))
 	return []func(next http.Handler) http.Handler{
 		authenticateMiddleware,
 		otelMiddleware,

internal/pkg/app/middlewares_test.go

@@ -87,7 +87,7 @@ func TestAuthMiddleware(t *testing.T) {
 
 	w := httptest.NewRecorder()
 	r := httptest.NewRequest(http.MethodGet, "/", nil)
-	r.Header.Set("X-B3-TraceId", "123456")
+	r.Header.Set(HeaderTraceId, "123456")
 
 	m.ServeHTTP(w, r.WithContext(subCtx))
 }

internal/pkg/appsecret/config_vault.go

@@ -14,7 +14,7 @@ import (
 	"time"
 )
 
-//ConfigVault provides secrets from ears app configuration
+// ConfigVault provides secrets from ears app configuration
 type ConfigVault struct {
 	config config.Config
 }
@@ -139,13 +139,19 @@ func (v *TenantConfigVault) getSatBearerToken(ctx context.Context) (string, erro
 	if err != nil {
 		return "", err
 	}
-	if len(tenantConfig.ClientIds) == 0 {
+	clientId := ""
+	if tenantConfig.ClientId != "" {
+		clientId = tenantConfig.ClientId
+	} else if len(tenantConfig.ClientIds) > 0 {
+		clientId = tenantConfig.ClientIds[0]
+	}
+	if clientId == "" {
 		return "", errors.New("tenant has no client IDs")
 	}
 	if tenantConfig.ClientSecret == "" {
 		return "", errors.New("tenant has no client secret")
 	}
-	req.Header.Add("X-Client-Id", tenantConfig.ClientIds[0])
+	req.Header.Add("X-Client-Id", clientId)
 	req.Header.Add("X-Client-Secret", tenantConfig.ClientSecret)
 	req.Header.Add("Cache-Control", "no-cache")
 	resp, err := v.httpClient.Do(req)

internal/pkg/tablemgr/routing_table_manager.go

@@ -227,9 +227,8 @@ func (r *DefaultRoutingTableManager) RouteEvent(ctx context.Context, tid tenant.
 	}
 	var wg sync.WaitGroup
 	wg.Add(1)
-	//sctx, cancel := context.WithTimeout(ctx, 5*time.Second)
-	//sctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	// no need to cancel context here because RouteEvent is only used synchronously via API call
+	//sctx, cancel := context.WithTimeout(ctx, 5*time.Second)
 	e, err := event.New(ctx, payload, event.WithAck(
 		func(evt event.Event) {
 			r.logger.Info().Str("op", "routeEventWebhook").Str("tid", tid.ToString()).Msg("success")
@@ -246,10 +245,10 @@ func (r *DefaultRoutingTableManager) RouteEvent(ctx context.Context, tid tenant.
 		event.WithTracePayloadOnNack(false),
 	)
 	if err != nil {
-		return "", errors.New("bad test event for route " + routeId)
+		return "", errors.New("invalid event for route " + routeId)
 	}
-	traceId, _, _ := e.GetPathValue("trace.id")
-	traceIdStr, _ := traceId.(string)
+	actualTraceId, _, _ := e.GetPathValue("trace.id")
+	traceIdStr, _ := actualTraceId.(string)
 	lrw.Receiver.Trigger(e)
 	wg.Wait()
 	return traceIdStr, nil

pkg/app/types.go

@@ -0,0 +1,20 @@
+// Copyright 2021 Comcast Cable Communications Management, LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package app
+
+const (
+	HeaderTraceId  = "X-B3-TraceId"
+	HeaderTenantId = "Application-Id"
+)