feat/ingestion rules v1 APIs

ee/query-service/app/api/api.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/gorilla/mux"
 	"go.signoz.io/signoz/ee/query-service/dao"
+	"go.signoz.io/signoz/ee/query-service/ingestionRules"
 	"go.signoz.io/signoz/ee/query-service/interfaces"
 	"go.signoz.io/signoz/ee/query-service/license"
 	baseapp "go.signoz.io/signoz/pkg/query-service/app"
@@ -15,11 +16,13 @@ import (
 )
 
 type APIHandlerOptions struct {
-	DataConnector  interfaces.DataConnector
-	AppDao         dao.ModelDao
-	RulesManager   *rules.Manager
-	FeatureFlags   baseint.FeatureLookup
-	LicenseManager *license.Manager
+	DataConnector       interfaces.DataConnector
+	AppDao              dao.ModelDao
+	RulesManager        *rules.Manager
+	FeatureFlags        baseint.FeatureLookup
+	LicenseManager      *license.Manager
+	Authenticator       *baseapp.AuthMiddleware
+	IngestionController *ingestionRules.IngestionController
 }
 
 type APIHandler struct {
@@ -31,10 +34,11 @@ type APIHandler struct {
 func NewAPIHandler(opts APIHandlerOptions) (*APIHandler, error) {
 
 	baseHandler, err := baseapp.NewAPIHandler(baseapp.APIHandlerOpts{
-		Reader:       opts.DataConnector,
-		AppDao:       opts.AppDao,
-		RuleManager:  opts.RulesManager,
-		FeatureFlags: opts.FeatureFlags})
+		Reader:        opts.DataConnector,
+		AppDao:        opts.AppDao,
+		RuleManager:   opts.RulesManager,
+		FeatureFlags:  opts.FeatureFlags,
+		Authenticator: opts.Authenticator})
 
 	if err != nil {
 		return nil, err
@@ -59,6 +63,10 @@ func (ah *APIHandler) LM() *license.Manager {
 	return ah.opts.LicenseManager
 }
 
+func (ah *APIHandler) Auth() *baseapp.AuthMiddleware {
+	return ah.opts.Authenticator
+}
+
 func (ah *APIHandler) AppDao() dao.ModelDao {
 	return ah.opts.AppDao
 }
@@ -69,66 +77,89 @@ func (ah *APIHandler) CheckFeature(f string) bool {
 }
 
 // RegisterRoutes registers routes for this handler on the given router
-func (ah *APIHandler) RegisterRoutes(router *mux.Router, am *baseapp.AuthMiddleware) {
+func (ah *APIHandler) RegisterRoutes(router *mux.Router) {
 	// note: add ee override methods first
 
 	// routes available only in ee version
 	router.HandleFunc("/api/v1/licenses",
-		am.AdminAccess(ah.listLicenses)).
+		ah.Auth().AdminAccess(ah.listLicenses)).
 		Methods(http.MethodGet)
 
 	router.HandleFunc("/api/v1/licenses",
-		am.AdminAccess(ah.applyLicense)).
+		ah.Auth().AdminAccess(ah.applyLicense)).
 		Methods(http.MethodPost)
 
 	router.HandleFunc("/api/v1/featureFlags",
-		am.OpenAccess(ah.getFeatureFlags)).
+		ah.Auth().OpenAccess(ah.getFeatureFlags)).
 		Methods(http.MethodGet)
 
 	router.HandleFunc("/api/v1/loginPrecheck",
-		am.OpenAccess(ah.precheckLogin)).
+		ah.Auth().OpenAccess(ah.precheckLogin)).
 		Methods(http.MethodGet)
 
 	// paid plans specific routes
 	router.HandleFunc("/api/v1/complete/saml",
-		am.OpenAccess(ah.receiveSAML)).
+		ah.Auth().OpenAccess(ah.receiveSAML)).
 		Methods(http.MethodPost)
 
 	router.HandleFunc("/api/v1/complete/google",
-		am.OpenAccess(ah.receiveGoogleAuth)).
+		ah.Auth().OpenAccess(ah.receiveGoogleAuth)).
 		Methods(http.MethodGet)
 
 	router.HandleFunc("/api/v1/orgs/{orgId}/domains",
-		am.AdminAccess(ah.listDomainsByOrg)).
+		ah.Auth().AdminAccess(ah.listDomainsByOrg)).
 		Methods(http.MethodGet)
 
 	router.HandleFunc("/api/v1/domains",
-		am.AdminAccess(ah.postDomain)).
+		ah.Auth().AdminAccess(ah.postDomain)).
 		Methods(http.MethodPost)
 
 	router.HandleFunc("/api/v1/domains/{id}",
-		am.AdminAccess(ah.putDomain)).
+		ah.Auth().AdminAccess(ah.putDomain)).
 		Methods(http.MethodPut)
 
 	router.HandleFunc("/api/v1/domains/{id}",
-		am.AdminAccess(ah.deleteDomain)).
+		ah.Auth().AdminAccess(ah.deleteDomain)).
 		Methods(http.MethodDelete)
 
+	router.HandleFunc("/api/v1/dropRules/{version}",
+		ah.Auth().AdminAccess(ah.listDropRules)).
+		Methods(http.MethodGet)
+
+	router.HandleFunc("/api/v1/dropRules/{version}/deploy",
+		ah.Auth().AdminAccess(ah.deployDropRules)).
+		Methods(http.MethodPost)
+
+	router.HandleFunc("/api/v1/dropRules",
+		ah.Auth().AdminAccess(ah.createDropRule)).
+		Methods(http.MethodPost)
+
+	router.HandleFunc("/api/v1/samplingRules/{version}/deploy",
+		ah.Auth().AdminAccess(ah.deploySamplingRules)).
+		Methods(http.MethodPost)
+
+	router.HandleFunc("/api/v1/samplingRules/{version}",
+		ah.Auth().AdminAccess(ah.listSamplingRules)).
+		Methods(http.MethodGet)
+
+	router.HandleFunc("/api/v1/samplingRules",
+		ah.Auth().AdminAccess(ah.createSamplingRule)).
+		Methods(http.MethodPost)
+
 	// base overrides
-	router.HandleFunc("/api/v1/version", am.OpenAccess(ah.getVersion)).Methods(http.MethodGet)
-	router.HandleFunc("/api/v1/invite/{token}", am.OpenAccess(ah.getInvite)).Methods(http.MethodGet)
-	router.HandleFunc("/api/v1/register", am.OpenAccess(ah.registerUser)).Methods(http.MethodPost)
-	router.HandleFunc("/api/v1/login", am.OpenAccess(ah.loginUser)).Methods(http.MethodPost)
-	router.HandleFunc("/api/v1/traces/{traceId}", am.ViewAccess(ah.searchTraces)).Methods(http.MethodGet)
-	router.HandleFunc("/api/v2/metrics/query_range", am.ViewAccess(ah.queryRangeMetricsV2)).Methods(http.MethodPost)
+	router.HandleFunc("/api/v1/version", ah.Auth().OpenAccess(ah.getVersion)).Methods(http.MethodGet)
+	router.HandleFunc("/api/v1/invite/{token}", ah.Auth().OpenAccess(ah.getInvite)).Methods(http.MethodGet)
+	router.HandleFunc("/api/v1/register", ah.Auth().OpenAccess(ah.registerUser)).Methods(http.MethodPost)
+	router.HandleFunc("/api/v1/login", ah.Auth().OpenAccess(ah.loginUser)).Methods(http.MethodPost)
+	router.HandleFunc("/api/v1/traces/{traceId}", ah.Auth().ViewAccess(ah.searchTraces)).Methods(http.MethodGet)
+	router.HandleFunc("/api/v2/metrics/query_range", ah.Auth().ViewAccess(ah.queryRangeMetricsV2)).Methods(http.MethodPost)
 
 	// PAT APIs
-	router.HandleFunc("/api/v1/pat", am.OpenAccess(ah.createPAT)).Methods(http.MethodPost)
-	router.HandleFunc("/api/v1/pat", am.OpenAccess(ah.getPATs)).Methods(http.MethodGet)
-	router.HandleFunc("/api/v1/pat/{id}", am.OpenAccess(ah.deletePAT)).Methods(http.MethodDelete)
-
-	ah.APIHandler.RegisterRoutes(router, am)
+	router.HandleFunc("/api/v1/pat", ah.Auth().OpenAccess(ah.createPAT)).Methods(http.MethodPost)
+	router.HandleFunc("/api/v1/pat", ah.Auth().OpenAccess(ah.getPATs)).Methods(http.MethodGet)
+	router.HandleFunc("/api/v1/pat/{id}", ah.Auth().OpenAccess(ah.deletePAT)).Methods(http.MethodDelete)
 
+	ah.APIHandler.RegisterRoutes(router)
 }
 
 func (ah *APIHandler) getVersion(w http.ResponseWriter, r *http.Request) {

ee/query-service/app/api/dropRules.go

@@ -0,0 +1,19 @@
+package api
+
+import (
+	"net/http"
+
+	"go.signoz.io/signoz/pkg/query-service/agentConf"
+)
+
+func (ah *APIHandler) listDropRules(w http.ResponseWriter, r *http.Request) {
+	ah.listIngestionRulesHandler(w, r, agentConf.ElementTypeSamplingRules)
+}
+
+func (ah *APIHandler) createDropRule(w http.ResponseWriter, r *http.Request) {
+	ah.createIngestionRule(w, r, agentConf.ElementTypeDropRules)
+}
+
+func (ah *APIHandler) deployDropRules(w http.ResponseWriter, r *http.Request) {
+	ah.redeployIngestionRule(w, r, agentConf.ElementTypeDropRules)
+}

ee/query-service/app/api/ingestionRules.go

@@ -0,0 +1,161 @@
+package api
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"strconv"
+
+	"github.com/gorilla/mux"
+	"go.signoz.io/signoz/ee/query-service/ingestionRules"
+	"go.signoz.io/signoz/ee/query-service/model"
+	"go.signoz.io/signoz/pkg/query-service/agentConf"
+	"go.uber.org/zap"
+)
+
+// ingestion rules handler - combines common methods for drop and sampling rules
+
+// parseAgentConfigVersion parses the version string from URL path.
+func parseAgentConfigVersion(r *http.Request) (int, *model.ApiError) {
+	// parse version from path and respond with
+	// - (-1) to indicate no version specified in the url path but the user is looking for the latest version
+	// - (0) indicates a failure in parsing
+	// - non-zero (>0) to indicate user is looking for a specific version
+
+	versionString := mux.Vars(r)["version"]
+
+	if versionString == "latest" {
+		return -1, nil
+	}
+
+	version64, err := strconv.ParseInt(versionString, 0, 8)
+
+	if err != nil {
+		return 0, model.BadRequestStr("invalid version number")
+	}
+
+	return int(version64), nil
+}
+
+func (ah *APIHandler) listIngestionRulesHandler(w http.ResponseWriter, r *http.Request, elementType agentConf.ElementTypeDef) {
+
+	version, err := parseAgentConfigVersion(r)
+	if err != nil {
+		RespondError(w, model.BadRequestStr("invalid version"), nil)
+		return
+	}
+
+	var payload *ingestionRules.IngestionRulesResponse
+	var apierr *model.ApiError
+
+	if version > 0 {
+		payload, apierr = ah.listIngestionRulesByVersion(context.Background(), version, elementType)
+	} else {
+		payload, apierr = ah.listIngestionRules(context.Background(), elementType)
+	}
+
+	if apierr != nil {
+		RespondError(w, apierr, payload)
+		return
+	}
+	ah.Respond(w, payload)
+}
+
+// listIngestionRules lists rules for latest version
+func (ah *APIHandler) listIngestionRules(ctx context.Context, elementType agentConf.ElementTypeDef) (*ingestionRules.IngestionRulesResponse, *model.ApiError) {
+
+	// get lateset agent config
+	lastestConfig, err := agentConf.GetLatestVersion(ctx, elementType)
+	if err != nil || lastestConfig == nil {
+		zap.S().Errorf("failed to get latest agent config version ", err)
+		return nil, model.InternalErrorStr("Failed to get latest agent config version")
+	}
+
+	payload, apierr := ah.opts.IngestionController.GetRulesByVersion(ctx, lastestConfig.Version, elementType)
+	if apierr != nil {
+		return payload, apierr
+	}
+
+	history, err := agentConf.GetConfigHistory(ctx, elementType, 10)
+	if apierr != nil {
+		return payload, apierr
+	}
+	payload.History = history
+	return payload, nil
+}
+
+// listIngestionRulesByVersion lists rules along with config version history
+func (ah *APIHandler) listIngestionRulesByVersion(ctx context.Context, version int, elementType agentConf.ElementTypeDef) (*ingestionRules.IngestionRulesResponse, *model.ApiError) {
+
+	payload, apierr := ah.opts.IngestionController.GetRulesByVersion(ctx, version, elementType)
+	if apierr != nil {
+		return payload, apierr
+	}
+
+	history, err := agentConf.GetConfigHistory(ctx, elementType, 10)
+	if err != nil {
+		zap.S().Errorf("failed to retreive config history for element type", elementType, err)
+		return payload, model.InternalErrorStr("failed to retrieve agent config history")
+	}
+
+	payload.History = history
+	return payload, nil
+}
+
+func (ah *APIHandler) createIngestionRule(w http.ResponseWriter, r *http.Request, elementType agentConf.ElementTypeDef) {
+	ctx := context.Background()
+	userPayload, err := ah.Auth().GetUserFromRequest(r)
+	if err != nil {
+		RespondError(w, model.BadRequestStr("failed to identify user from the request"), nil)
+	}
+
+	req := ingestionRules.PostableIngestionRules{}
+
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		RespondError(w, model.BadRequest(err), nil)
+		return
+	}
+
+	createRule := func(ctx context.Context, userId string, postable []ingestionRules.PostableIngestionRule) (*ingestionRules.IngestionRulesResponse, *model.ApiError) {
+		if len(postable) == 0 {
+			zap.S().Warnf("found no rules in the http request, this will delete all the rules")
+		}
+
+		for _, p := range postable {
+			if apierr := p.IsValid(); apierr != nil {
+				zap.S().Debugf("received invalid dropping rule in the POST request", apierr)
+				return nil, apierr
+			}
+		}
+
+		return ah.opts.IngestionController.ApplyRules(ctx, userId, elementType, postable)
+	}
+
+	ingestionRuleResponse, apierr := createRule(ctx, userPayload.User.Id, req.Rules)
+	if apierr != nil {
+		RespondError(w, apierr, nil)
+		return
+	}
+
+	ah.Respond(w, ingestionRuleResponse)
+}
+
+func (ah *APIHandler) redeployIngestionRule(w http.ResponseWriter, r *http.Request, elementType agentConf.ElementTypeDef) {
+	version, apierr := parseAgentConfigVersion(r)
+	if apierr != nil {
+		RespondError(w, apierr, nil)
+		return
+	}
+
+	if version == 0 {
+		RespondError(w, model.BadRequestStr("config version required"), nil)
+		return
+	}
+
+	if err := agentConf.Redeploy(context.Background(), elementType, version); err != nil {
+		RespondError(w, model.InternalError(err), nil)
+		return
+	}
+
+	ah.Respond(w, "deployment started")
+}

ee/query-service/app/api/pat.go

@@ -25,6 +25,12 @@ func generatePATToken() string {
 }
 
 func (ah *APIHandler) createPAT(w http.ResponseWriter, r *http.Request) {
+
+	if !ah.CheckFeature(model.Pat) {
+		RespondError(w, model.BadRequestStr("feature unavailable, please upgrade to EE"), nil)
+		return
+	}
+
 	ctx := context.Background()
 
 	req := model.PAT{}
@@ -57,6 +63,11 @@ func (ah *APIHandler) createPAT(w http.ResponseWriter, r *http.Request) {
 }
 
 func (ah *APIHandler) getPATs(w http.ResponseWriter, r *http.Request) {
+	if !ah.CheckFeature(model.Pat) {
+		RespondError(w, model.BadRequestStr("feature unavailable, please upgrade to EE"), nil)
+		return
+	}
+
 	ctx := context.Background()
 	user, err := auth.GetUserFromRequest(r)
 	if err != nil {
@@ -76,6 +87,11 @@ func (ah *APIHandler) getPATs(w http.ResponseWriter, r *http.Request) {
 }
 
 func (ah *APIHandler) deletePAT(w http.ResponseWriter, r *http.Request) {
+	if !ah.CheckFeature(model.Pat) {
+		RespondError(w, model.BadRequestStr("feature unavailable, please upgrade to EE"), nil)
+		return
+	}
+
 	ctx := context.Background()
 	id := mux.Vars(r)["id"]
 	user, err := auth.GetUserFromRequest(r)