Backlog/v12 integation index patterns

backend/modules.go

@@ -162,11 +162,6 @@ func initModules(db *gorm.DB, cfg *config) *modules {
 	}
 	datasourcesMod := datasources.NewModule(dsUC, dsGroupUC, dsReconciler, billingMod.License(), agentClient)
 
-	integrationsMod := integrations.NewModule(db, cipher,
-		env.String("INTEGRATIONS_TENANT_DIR", "/workdir/pipeline", false),
-		dsUC,
-	)
-
 	opensearchMod := opensearchgw.NewModule(db, cfg.esHost != "")
 	notificationsMod := notifications.NewModule(db, auditMod.Logger())
 
@@ -184,6 +179,7 @@ func initModules(db *gorm.DB, cfg *config) *modules {
 		)
 	}
 
+
 	iamMod := iam.NewModule(authUsecase, userUsecase, roleUsecase, tfaUsecase, apiKeyUsecase, idpUsecase, samlUsecase, cfg.uploadDir)
 	socAIMod := socai.NewModule(cfg.socAIBaseURL, cfg.internalKey, cipher,
 		env.String("INTEGRATIONS_TENANT_DIR", "/workdir/pipeline", false))
@@ -196,6 +192,13 @@ func initModules(db *gorm.DB, cfg *config) *modules {
 	)
 	adauditMod := adaudit.NewModule(db)
 
+	//loaded after opensearch has fully loaded
+	integrationsMod := integrations.NewModule(db, cipher,
+		env.String("INTEGRATIONS_TENANT_DIR", "/workdir/pipeline", false),
+		dsUC,
+		opensearchMod.GetIndexPatternUsecase(),
+	)
+
 	var mcpModule *mcpmod.Module
 	if cfg.mcpEnabled {
 		mcpModule = mcpmod.NewModule(&mcpmod.Deps{

backend/modules/integrations/module.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/threatwinds/go-sdk/catcher"
 	ds_connectors "github.com/utmstack/utmstack/backend/modules/datasources/connectors"
+	os_connectors "github.com/utmstack/utmstack/backend/modules/opensearch/connectors"
 	"github.com/utmstack/utmstack/backend/modules/integrations/connectors"
 	"github.com/utmstack/utmstack/backend/modules/integrations/repository"
 	"github.com/utmstack/utmstack/backend/modules/integrations/usecase"
@@ -17,16 +18,23 @@ type Module struct {
 	tenants   *usecase.TenantUsecase
 	modules   connectors.ModuleUsecase
 	bootstrap *usecase.Bootstrap
+	opensearch os_connectors.IndexPatternUsecase
 }
 
-func NewModule(db *gorm.DB, cipher connectors.Cipher, tenantDir string, datasources ds_connectors.DatasourceUsecase) *Module {
+func NewModule(
+	db *gorm.DB,
+	cipher connectors.Cipher,
+	tenantDir string,
+	datasources ds_connectors.DatasourceUsecase,
+	opensearch os_connectors.IndexPatternUsecase,
+) *Module {
 	store := repository.NewTenantStore(tenantDir)
 	schema := repository.NewCodeSchemaProvider() // field schema lives in code
 	verif := verifier.NewBackendVerifier()
 
 	moduleRepo := repository.NewModuleRepository(db)
 	// store doubles as the tenant-file toggler: enable/disable <module>.yaml(.disabled).
-	moduleUC := usecase.NewModuleUsecase(moduleRepo, store)
+	moduleUC := usecase.NewModuleUsecase(moduleRepo, store,opensearch)
 
 	return &Module{
 		db:        db,

backend/modules/integrations/usecase/module.go

@@ -3,24 +3,30 @@ package usecase
 import (
 	"context"
 	"fmt"
-
+	"github.com/threatwinds/go-sdk/catcher"
 	"github.com/utmstack/utmstack/backend/modules/integrations/connectors"
 	"github.com/utmstack/utmstack/backend/modules/integrations/domain"
 	"github.com/utmstack/utmstack/backend/modules/integrations/dto"
+
+	os_connectors "github.com/utmstack/utmstack/backend/modules/opensearch/connectors"
+	os_dto "github.com/utmstack/utmstack/backend/modules/opensearch/dto"
 )
 
 type moduleUsecase struct {
 	repo          connectors.ModuleRepository
 	tenantFileTog connectors.TenantFileToggler
+	opensearch    os_connectors.IndexPatternUsecase
 }
 
 func NewModuleUsecase(
 	repo connectors.ModuleRepository,
 	tenantFile connectors.TenantFileToggler,
+	opensearch os_connectors.IndexPatternUsecase,
 ) connectors.ModuleUsecase {
 	return &moduleUsecase{
 		repo:          repo,
 		tenantFileTog: tenantFile,
+		opensearch: opensearch,
 	}
 }
 
@@ -51,6 +57,32 @@ func (u *moduleUsecase) ActivateDeactivate(ctx context.Context, req dto.ModuleAc
 		return nil, err
 	}
 
+	results,count,err :=u.opensearch.List(ctx,os_dto.IndexPatternFilters{
+		ModuleEquals: &module.ModuleName,
+	})
+	if err != nil {
+		return nil,err
+	}
+	if count > 0 && len(results)>0 {
+		//if module has index patterns
+		pattern:=results[0]
+
+		if _,err := u.opensearch.Update(ctx,os_dto.UpdateIndexPatternRequest{
+			ID: pattern.ID,
+			Pattern: pattern.Pattern,
+			PatternModule: pattern.PatternModule,
+			IsActive: &fresh.ModuleActive,
+			PatternSystem:pattern.PatternSystem,
+		});err!=nil{
+			catcher.Error("failed to toggle is active for module's index pattern, rolling back", err, map[string]any{"module":module.ModuleName,"pattern":pattern} )
+			module.ModuleActive = !active
+			if err_ := u.repo.Save(ctx, module); err_ != nil {
+				catcher.Error("failed to change module status after index pattern activation toggle", err, map[string]any{"module":module.ModuleName,"pattern":pattern} )
+				return nil, err
+			}
+			return nil,err
+		}
+	}
 	resp := dto.FromModule(*fresh)
 	return &resp, nil
 }
@@ -128,9 +160,22 @@ func (u *moduleUsecase) Create(ctx context.Context, req dto.CreateModuleRequest)
 		ModuleActive:      false,
 		IsSystem:          false,
 	}
+
+
 	if err := u.repo.Save(ctx, m); err != nil {
 		return nil, err
 	}
+
+	if _,err:= u.opensearch.Create(ctx,os_dto.CreateIndexPatternRequest{
+		PatternModule: &req.ModuleName,
+	});err!=nil{
+		//opensearch insertion fails, rolling back database
+		if err:=u.repo.Delete(ctx,m.ID);err!=nil{
+		  catcher.Error("failed to remove module after index pattern creation fail", err, map[string]any{"module":m.ModuleName} )
+		}
+		return nil, err
+	}
+
 	resp := dto.FromModule(*m)
 	return &resp, nil
 }
@@ -163,5 +208,30 @@ func (u *moduleUsecase) Delete(ctx context.Context, id int64) error {
 	if m.IsSystem {
 		return domain.ErrSystemModule
 	}
-	return u.repo.Delete(ctx, id)
+
+	results,count,err :=u.opensearch.List(ctx,os_dto.IndexPatternFilters{
+		ModuleEquals: &m.ModuleName,
+	})
+	if err != nil {
+		return err
+	}
+	if count == 0 || len(results)==0{
+		return fmt.Errorf("module %s index pattern not found.",m.ModuleName)
+	}
+
+	err= u.repo.Delete(ctx, id)
+	if err!=nil{
+		return err
+	}
+
+	pattern:=results[0]
+	if err:=u.opensearch.Delete(ctx,pattern.ID);err!=nil{
+		//if index patter deletion fails integration must be restored
+		if err:=u.repo.Save(ctx,m);err!=nil{
+			catcher.Error("failed to restore module after index pattern remove fail", err, map[string]any{"module":m.ModuleName} )
+		}
+		return err
+	}
+
+	return nil
 }

backend/modules/opensearch/connectors/repository.go

@@ -12,6 +12,7 @@ type GatewayRepository interface {
 	PropertyValuesWithCount(ctx context.Context, index, field string, filters map[string]any, top int, orderByCount, sortAsc bool) (map[string]int64, error)
 	IndexProperties(ctx context.Context, pattern string) ([]dto.IndexPropertyType, error)
 	Indices(ctx context.Context, pattern string) ([]dto.IndexInfo, error)
+	EnsureIndexPattern(ctx context.Context, pattern string) error
 	DeleteIndices(ctx context.Context, indices []string) error
 	Search(ctx context.Context, index string, query map[string]any, from, size int, sortField, sortOrder string) ([]map[string]any, int64, error)
 	Count(ctx context.Context, index string, query map[string]any) (int64, error)

backend/modules/opensearch/repository/index_pattern_pg.go

@@ -19,6 +19,11 @@ func NewIndexPatternRepository(db *gorm.DB) connectors.IndexPatternRepository {
 }
 
 func (r *pgIndexPatternRepository) Create(ctx context.Context, p *domain.UtmIndexPattern) error {
+	sanitized, err := ensureIndexPattern(ctx, p.Pattern)
+	if err != nil {
+		return err
+	}
+	p.Pattern = sanitized
 	return r.db.WithContext(ctx).Create(p).Error
 }
 

backend/modules/opensearch/repository/opensearch_gateway.go

@@ -3,16 +3,33 @@ package repository
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
+	"net/http"
+	"regexp"
 	"sort"
 	"strconv"
 	"strings"
 
 	osdk "github.com/threatwinds/go-sdk/os"
 	"github.com/utmstack/utmstack/backend/modules/opensearch/connectors"
 	"github.com/utmstack/utmstack/backend/modules/opensearch/dto"
+	"github.com/utmstack/utmstack/backend/pkg/constants"
 )
 
+var (
+	reInvalidPatternChars = regexp.MustCompile(`[^a-z0-9_\-*]+`)
+	reCollapsePatternDash = regexp.MustCompile(`-+`)
+)
+
+func sanitizePattern(pattern string) string {
+	s := strings.ToLower(strings.TrimSpace(pattern))
+	s = reInvalidPatternChars.ReplaceAllString(s, "-")
+	s = reCollapsePatternDash.ReplaceAllString(s, "-")
+	s = strings.Trim(s, "-_")
+	return s
+}
+
 type osGatewayRepo struct {
 	mapper *osdk.FieldMapper
 }
@@ -135,6 +152,60 @@ func (r *osGatewayRepo) Indices(ctx context.Context, pattern string) ([]dto.Inde
 	return out, nil
 }
 
+func (r *osGatewayRepo) EnsureIndexPattern(ctx context.Context, pattern string) error {
+	_, err := ensureIndexPattern(ctx, pattern)
+	return err
+}
+
+func ensureIndexPattern(ctx context.Context, pattern string) (string, error) {
+	sanitized := sanitizePattern(pattern)
+	if sanitized == "" || sanitized == "*" {
+		return "", errors.New("opensearch: pattern is empty")
+	}
+	name := templateNameFor(sanitized)
+	if name == "" {
+		return "", fmt.Errorf("opensearch: pattern %q has no valid template name", sanitized)
+	}
+	path := "_index_template/" + name
+
+	_, status, err := osdk.DoRequest(ctx, http.MethodGet, path, nil)
+	if err != nil {
+		return "", err
+	}
+	if status == http.StatusOK {
+		return sanitized, nil
+	}
+	if status != http.StatusNotFound {
+		return "", fmt.Errorf("opensearch: probe %q: status %d", path, status)
+	}
+
+	body := map[string]any{
+		"index_patterns": []string{sanitized},
+		"priority":       0,
+		"template": map[string]any{
+			"settings": map[string]any{
+				"index.mapping.total_fields.limit": constants.OS_INDEX_FIELD_LIMIT,
+				"number_of_shards":                 constants.OS_INDEX_NUMBER_OF_SHARDS,
+				"number_of_replicas":               constants.OS_INDEX_NUMBER_OF_REPLICAS,
+			},
+		},
+	}
+	data, status, err := osdk.DoRequest(ctx, http.MethodPut, path, body)
+	if err != nil {
+		return "", err
+	}
+	if status < 200 || status >= 300 {
+		return "", fmt.Errorf("opensearch: create index pattern %q: status %d: %s", sanitized, status, string(data))
+	}
+	return sanitized, nil
+}
+
+func templateNameFor(pattern string) string {
+	n := strings.ReplaceAll(pattern, "*", "")
+	n = strings.Trim(n, "-_.")
+	return n
+}
+
 func (r *osGatewayRepo) DeleteIndices(ctx context.Context, indices []string) error {
 	for _, idx := range indices {
 		if err := osdk.DeleteIndex(ctx, idx); err != nil {

backend/modules/opensearch/repository/opensearch_ism.go

@@ -9,6 +9,7 @@ import (
 	osdk "github.com/threatwinds/go-sdk/os"
 	"github.com/utmstack/utmstack/backend/modules/opensearch/domain"
 	"github.com/utmstack/utmstack/backend/modules/opensearch/dto"
+	"github.com/utmstack/utmstack/backend/pkg/constants"
 )
 
 // ISMClient performs Index State Management (and snapshot/mapping) calls against
@@ -78,8 +79,6 @@ func (c *ISMClient) UpdatePolicy(ctx context.Context, policy domain.IndexPolicy,
 	return nil
 }
 
-const fieldLimit = 50000
-
 func (c *ISMClient) EnsureFieldLimitTemplate(ctx context.Context, patterns []string) error {
 	if len(patterns) == 0 {
 		return nil
@@ -90,9 +89,9 @@ func (c *ISMClient) EnsureFieldLimitTemplate(ctx context.Context, patterns []str
 		"priority":       0,
 		"template": map[string]any{
 			"settings": map[string]any{
-				"index.mapping.total_fields.limit": fieldLimit,
-				"number_of_shards":                 3,
-				"number_of_replicas":               0,
+				"index.mapping.total_fields.limit": constants.OS_INDEX_FIELD_LIMIT,
+				"number_of_shards":                 constants.OS_INDEX_NUMBER_OF_SHARDS,
+				"number_of_replicas":               constants.OS_INDEX_NUMBER_OF_REPLICAS,
 			},
 		},
 	}
@@ -108,7 +107,7 @@ func (c *ISMClient) EnsureFieldLimitTemplate(ctx context.Context, patterns []str
 
 func (c *ISMClient) RaiseFieldLimit(ctx context.Context, pattern string) error {
 	path := fmt.Sprintf("%s/_settings", pattern)
-	body := map[string]any{"index.mapping.total_fields.limit": fieldLimit}
+	body := map[string]any{"index.mapping.total_fields.limit": constants.OS_INDEX_FIELD_LIMIT}
 	data, status, err := c.do(ctx, http.MethodPut, path, body)
 	if err != nil {
 		return err

backend/modules/opensearch/usecase/index_pattern.go

@@ -3,14 +3,18 @@ package usecase
 import (
 	"context"
 	"errors"
+	"fmt"
+	"strings"
 
 	"github.com/threatwinds/go-sdk/catcher"
 	"github.com/utmstack/utmstack/backend/modules/opensearch/connectors"
 	"github.com/utmstack/utmstack/backend/modules/opensearch/domain"
 	"github.com/utmstack/utmstack/backend/modules/opensearch/dto"
+	"github.com/utmstack/utmstack/backend/pkg/constants"
 )
 
 var errIDMustBeAbsent = errors.New("id must be absent on create")
+var errInvalidPattern = errors.New("invalid pattern")
 
 type indexPatternUsecase struct {
 	repo    connectors.IndexPatternRepository
@@ -36,6 +40,15 @@ func (u *indexPatternUsecase) Create(ctx context.Context, req dto.CreateIndexPat
 	}
 	systemFalse := false
 
+	if req.Pattern == "" && req.PatternModule != nil {
+		normalized := strings.ReplaceAll(strings.Trim(strings.ToLower(*req.PatternModule), ""), " ", "-")
+		req.Pattern = fmt.Sprintf("%s-%s-*", constants.CUSTOM_INDEX_PREFIX, normalized)
+	}
+
+	if req.Pattern == "" || req.Pattern == " " {
+		return nil, errInvalidPattern
+	}
+
 	p := &domain.UtmIndexPattern{
 		Pattern:       req.Pattern,
 		PatternModule: req.PatternModule,

backend/pkg/constants/custom_integrations.go

@@ -0,0 +1,5 @@
+package constants
+
+const(
+	CUSTOM_INDEX_PREFIX = "custom";
+)

backend/pkg/constants/opensearch_index.go

@@ -0,0 +1,7 @@
+package constants
+
+const (
+	OS_INDEX_FIELD_LIMIT       = 50000
+	OS_INDEX_NUMBER_OF_SHARDS  = 3
+	OS_INDEX_NUMBER_OF_REPLICAS = 0
+)