Merge pull request #14 from flant/flant/refresh-index-patterns

Обновление существующих index-patterns

Author: uzhinskiy

.github/workflows/docker.yml

@@ -49,4 +49,4 @@ jobs:
           versionflags=-X 'main.appVersion=${{ github.ref_name }}'
         push: ${{ github.event_name != 'pull_request' }}
         tags: ${{ steps.meta.outputs.tags }}
-        labels: ${{ steps.meta.outputs.labels }}
+        labels: ${{ steps.meta.outputs.labels }}

.github/workflows/go.yml

@@ -35,4 +35,4 @@ jobs:
         generate_release_notes: true
         token: ${{ secrets.GITHUB_TOKEN }}
         files: |
-          build/*
+          build/*

ARCHITECTURE.md

@@ -567,6 +567,12 @@ osctl/
 
 ### 13. **indexpatterns** - Управление Kibana index patterns
 
+**Обновление паттернов**
+Выполняется до создания паттернов. 
+1. Получаем список существующих index patterns - имя и ID
+2. Через запрос в API OpenSearchDasboard `GET /api/index_patterns/_fields_for_wildcard?pattern={ИМЯ}&meta_fields=_source&meta_fields=_id&meta_fields=_type&meta_fields=_index&meta_fields=_score` получаем актуальный маппинг для индекса
+3. Обновляем паттерн по его ID - `PUT /api/saved_objects/index-pattern/{ID}"`
+
 **Алгоритм (multitenancy):**
 
 1. **Загрузка конфигурации тенантов**: Получаем `--indexpatterns-kibana-tenants-config` (по умолчанию `osctltenants.yaml`)
@@ -605,6 +611,7 @@ osctl/
 - Использует `--indexpatterns-kibana-tenants-config` для пути к конфигу тенантов
 - Использует `--kibana-index-regex` для построения паттернов в single-tenant режиме
 - Использует `--indexpatterns-recoverer-enabled` для создания `extracted_*` pattern (только в single-tenant режиме)
+- Использует `--indexpatterns-refresh-enabled` для обновления существующих index patterns
 - `extracted_*` по соображениям безопасности не создается в режиме multitenancy
 
 ### 14. **datasource** - Создание Kibana data sources

ENVIRONMENT_VARIABLES_AND_FLAGS.md

@@ -51,9 +51,9 @@
 - `coldstorage` 
 - `extracteddelete`
 - `danglingchecker`
- - `sharding`
- - `indexpatterns`
- - `datasource`
+- `sharding`
+- `indexpatterns`
+- `datasource`
 
 ### Примеры использования:
 
@@ -181,6 +181,7 @@ osctl --action=snapshot
 | `--indexpatterns-kibana-multitenancy` | `INDEXPATTERNS_KIBANA_MULTITENANCY` | Режим multitenancy | `false` |
 | `--indexpatterns-kibana-tenants-config` | `INDEXPATTERNS_KIBANA_TENANTS_CONFIG` | Путь к YAML с тенантами и index patterns | `osctltenants.yaml` |
 | `--indexpatterns-recoverer-enabled` | `INDEXPATTERNS_RECOVERER_ENABLED` | Создавать `extracted_*` с ссылкой на data-source | `false` |
+| `--indexpatterns-refresh-enabled` | `INDEXPATTERNS_REFRESH_ENABLED` | Обновлять существующие index patterns| `false` |
 | `--dry-run` | `DRY_RUN` | Показать создаваемые index patterns без создания | `false` |
 
 Примечание: Список тенантов берётся из `--indexpatterns-kibana-tenants-config` (`INDEXPATTERNS_KIBANA_TENANTS_CONFIG`).

commands/indexpatterns.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"os"
 	"osctl/pkg/config"
+	"osctl/pkg/kibana"
 	"osctl/pkg/logging"
 	"osctl/pkg/opensearch"
 	"osctl/pkg/utils"
@@ -29,6 +30,9 @@ func init() {
 }
 
 func runIndexPatterns(cmd *cobra.Command, args []string) error {
+	var refreshedPatterns []string
+	var failedRefreshedPatterns []string
+
 	cfg := config.GetConfig()
 
 	if cfg.GetOSDURL() == "" {
@@ -40,6 +44,37 @@ func runIndexPatterns(cmd *cobra.Command, args []string) error {
 		return fmt.Errorf("failed to create OpenSearch client: %v", err)
 	}
 
+	if cfg.GetIndexPatternsRefreshEnabled() {
+
+		user := cfg.GetKibanaUser()
+		pass := cfg.GetKibanaPass()
+		_, _, existingIdsTitiles, err := getExistingIndexPatternTitles(osClient, ".kibana*")
+		if err != nil {
+			return err
+		}
+		kb := kibana.NewClient(utils.NormalizeURL(cfg.GetOSDURL()), user, pass, cfg.GetTimeout())
+		logger.Info(fmt.Sprintf("Will refresh %d existing index-patterns", len(existingIdsTitiles)))
+		for ip_id, ip_title := range existingIdsTitiles {
+			if ip_title == "*" {
+				logger.Warn("index-pattern '*' is too general and will be ignored")
+				continue
+			}
+			if cfg.GetDryRun() {
+				logger.Info(fmt.Sprintf("DRY RUN: Would refresh index-pattern %s:%s", ip_id, ip_title))
+				refreshedPatterns = append(refreshedPatterns, ip_title)
+			} else {
+				logger.Info(fmt.Sprintf("Refreshing index-pattern %s:%s", ip_id, ip_title))
+				if err := kb.RefreshIndexPattern(ip_id, ip_title); err == nil {
+					logger.Info(fmt.Sprintf("Successfully refreshed index-pattern %s:%s", ip_id, ip_title))
+					refreshedPatterns = append(refreshedPatterns, ip_title)
+				} else {
+					logger.Error(fmt.Sprintf("Failed to refresh index-pattern %s:%s: %s", ip_id, ip_title, err))
+					failedRefreshedPatterns = append(failedRefreshedPatterns, ip_title)
+				}
+			}
+		}
+
+	}
 	var createdPatterns []string
 
 	if cfg.GetIndexPatternsKibanaMultitenancy() {
@@ -68,7 +103,7 @@ func runIndexPatterns(cmd *cobra.Command, args []string) error {
 				logger.Info(fmt.Sprintf("Skip tenant %s: .kibana alias not found", t.Name))
 				continue
 			}
-			existing, existingTitles, err := getExistingIndexPatternTitles(osClient, tenantIndex)
+			existing, existingTitles, _, err := getExistingIndexPatternTitles(osClient, tenantIndex)
 			if err != nil {
 				return err
 			}
@@ -142,8 +177,9 @@ func runIndexPatterns(cmd *cobra.Command, args []string) error {
 				needed = append(needed, m[1]+"-*")
 			}
 		}
+
 		logger.Info(fmt.Sprintf("Required patterns (%d): %s", len(needed), strings.Join(needed, ", ")))
-		existing, existingTitles, err := getExistingIndexPatternTitles(osClient, ".kibana")
+		existing, existingTitles, _, err := getExistingIndexPatternTitles(osClient, ".kibana")
 		if err != nil {
 			return err
 		}
@@ -226,6 +262,23 @@ func runIndexPatterns(cmd *cobra.Command, args []string) error {
 		logger.Info(strings.Repeat("=", 60))
 		logger.Info("INDEX PATTERNS SUMMARY")
 		logger.Info(strings.Repeat("=", 60))
+
+		if len(refreshedPatterns) > 0 {
+			logger.Info(fmt.Sprintf("Refreshed: %d index patterns", len(refreshedPatterns)))
+			for _, name := range refreshedPatterns {
+				logger.Info(fmt.Sprintf("  ✓ %s", name))
+			}
+		} else {
+			logger.Info("No index patterns were refreshed")
+		}
+
+		if len(failedRefreshedPatterns) > 0 {
+			logger.Info(fmt.Sprintf("Failed while refresh: %d index patterns", len(failedRefreshedPatterns)))
+			for _, name := range failedRefreshedPatterns {
+				logger.Info(fmt.Sprintf("  ✕ %s", name))
+			}
+		}
+
 		if len(createdPatterns) > 0 {
 			logger.Info(fmt.Sprintf("Created: %d index patterns", len(createdPatterns)))
 			for _, name := range createdPatterns {
@@ -240,22 +293,28 @@ func runIndexPatterns(cmd *cobra.Command, args []string) error {
 	return nil
 }
 
-func getExistingIndexPatternTitles(osClient *opensearch.Client, index string) (map[string]struct{}, []string, error) {
+func getExistingIndexPatternTitles(osClient *opensearch.Client, index string) (map[string]struct{}, []string, map[string]string, error) {
 	sr, err := osClient.Search(index, "q=type:index-pattern&size=1000")
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, nil, err
 	}
 	existing := map[string]struct{}{}
+	exist_map_id_title := map[string]string{}
 	titles := []string{}
 	for _, h := range sr.Hits.Hits {
 		if src, ok := h.Source["index-pattern"].(map[string]any); ok {
 			if t, ok := src["title"].(string); ok {
 				if _, seen := existing[t]; !seen {
 					existing[t] = struct{}{}
 					titles = append(titles, t)
+
+				}
+				if _, seen := exist_map_id_title[h.ID]; !seen {
+					p_id := strings.Split(h.ID, ":")
+					exist_map_id_title[p_id[1]] = t
 				}
 			}
 		}
 	}
-	return existing, titles, nil
+	return existing, titles, exist_map_id_title, nil
 }

config.yaml

@@ -55,6 +55,7 @@ recoverer_date_format: "%d-%m-%Y"
 kibana_index_regex: '^([\w-]+)-([\w-]*)(\d{4}[\.-]\d{2}[\.-]\d{2}(?:[\.-]\d{2})*)$'
 indexpatterns_kibana_multitenancy: false
 indexpatterns_recoverer_enabled: true
+indexpatterns_refresh_enabled: false
 
 # indicesdelete:
 indicesdelete_check_snapshots: true

example/helm-example/templates/osctl/02-osctl-config.yaml

@@ -101,6 +101,7 @@ data:
     {{- else }}
     indexpatterns_recoverer_enabled: false
     {{- end }}
+    indexpatterns_refresh_enabled: {{ .Values.indexpatterns.refresh_enabled | default true | quote }}
 
     # indicesdelete:
     indicesdelete_check_snapshots: true

pkg/config/config.go

@@ -65,6 +65,7 @@ type Config struct {
 	IndexPatternsKibanaMultitenancy    string
 	IndexPatternsKibanaTenantsConfig   string
 	IndexPatternsRecovererEnabled      string
+	IndexPatternsRefreshEnabled        string
 	SnapshotsBackfillIndicesList       string
 	MaxConcurrentSnapshots             string
 }
@@ -181,6 +182,7 @@ func LoadConfig(cmd *cobra.Command, commandName string) error {
 		IndexPatternsKibanaMultitenancy:    getValue(cmd, "indexpatterns-kibana-multitenancy", "INDEXPATTERNS_KIBANA_MULTITENANCY", viper.GetString("indexpatterns_kibana_multitenancy")),
 		IndexPatternsKibanaTenantsConfig:   getValue(cmd, "indexpatterns-kibana-tenants-config", "INDEXPATTERNS_KIBANA_TENANTS_CONFIG", viper.GetString("indexpatterns_kibana_tenants_config")),
 		IndexPatternsRecovererEnabled:      getValue(cmd, "indexpatterns-recoverer-enabled", "INDEXPATTERNS_RECOVERER_ENABLED", viper.GetString("indexpatterns_recoverer_enabled")),
+		IndexPatternsRefreshEnabled:        getValue(cmd, "indexpatterns-refresh-enabled", "INDEXPATTERNS_REFRESH_ENABLED", viper.GetString("indexpatterns_refresh_enabled")),
 		SnapshotsBackfillIndicesList:       getValue(cmd, "indices-list", "SNAPSHOTS_BACKFILL_INDICES_LIST", viper.GetString("snapshots_backfill_indices_list")),
 		MaxConcurrentSnapshots:             getValue(cmd, "max-concurrent-snapshots", "MAX_CONCURRENT_SNAPSHOTS", viper.GetString("max_concurrent_snapshots")),
 	}
@@ -198,6 +200,12 @@ func LoadConfig(cmd *cobra.Command, commandName string) error {
 		if repoToUse == "" {
 			return fmt.Errorf("snap-repo is required (or set snapshot-manual-repo) for %s", commandName)
 		}
+	case "indexpatterns":
+		if parseBoolWithDefault(configInstance.IndexPatternsRefreshEnabled, "indexpatterns_refresh_enabled") {
+			if configInstance.KibanaUser == "" || configInstance.KibanaPass == "" {
+				return fmt.Errorf("kibana-user and kibana-pass are required when indexpatterns-refresh-enabled is true")
+			}
+		}
 	}
 
 	return nil
@@ -250,7 +258,8 @@ func setDefaults() {
 	viper.SetDefault("kibana_multidomain_enabled", false)
 	viper.SetDefault("datasource_name", "recoverer")
 	viper.SetDefault("datasource_endpoint", "https://opendistro-recoverer:9200")
-	viper.SetDefault("max_concurrent_snapshots", 3)
+	viper.SetDefault("indexpatterns_refresh_enabled", false)
+  viper.SetDefault("max_concurrent_snapshots", 3)
 }
 
 func GetAvailableActions() []string {
@@ -427,6 +436,10 @@ func (c *Config) GetIndexPatternsRecovererEnabled() bool {
 	return parseBoolWithDefault(c.IndexPatternsRecovererEnabled, "indexpatterns_recoverer_enabled")
 }
 
+func (c *Config) GetIndexPatternsRefreshEnabled() bool {
+	return parseBoolWithDefault(c.IndexPatternsRefreshEnabled, "indexpatterns_refresh_enabled")
+}
+
 func (c *Config) GetShardingTargetSizeGiB() int {
 	value := parseIntWithDefault(c.ShardingTargetSizeGiB, "sharding_target_size_gib")
 	if value < 1 || value > 50 {
@@ -665,6 +678,7 @@ var CommandFlags = map[string][]FlagDefinition{
 		{"indexpatterns-kibana-multitenancy", "bool", false, "Enable multitenancy mode", []string{}},
 		{"indexpatterns-kibana-tenants-config", "string", "osctltenants.yaml", "Path to YAML tenants and patterns", []string{}},
 		{"indexpatterns-recoverer-enabled", "bool", false, "Enable recoverer extracted_* pattern creation", []string{}},
+		{"indexpatterns-refresh-enabled", "bool", false, "Enable refreshing for existing index patterns", []string{}},
 		{"dry-run", "bool", false, "Show what index patterns would be created without creating", []string{}},
 	},
 	"snapshotsdelete": {

pkg/kibana/service.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"net/http"
 	"net/url"
+	"osctl/pkg/utils"
 	"strings"
 )
 
@@ -21,6 +22,10 @@ type FindResponse struct {
 	SavedObjects []SavedObject `json:"saved_objects"`
 }
 
+type IPFields struct {
+	Fields []map[string]any `json:"fields"`
+}
+
 func (c *Client) FindSavedObjects(tenant string, objType string, perPage int) (*FindResponse, error) {
 	params := url.Values{}
 	params.Set("type", objType)
@@ -68,7 +73,10 @@ func (c *Client) CreateDataSource(tenant, title, endpoint, user, password string
 			},
 		},
 	}
-	b, _ := json.Marshal(body)
+	b, err := json.Marshal(body)
+	if err != nil {
+		return fmt.Errorf("failed to marshal data source body: %w", err)
+	}
 	req, err := http.NewRequest("POST", u, bytes.NewReader(b))
 	if err != nil {
 		return err
@@ -88,3 +96,90 @@ func (c *Client) CreateDataSource(tenant, title, endpoint, user, password string
 	}
 	return nil
 }
+
+func (c *Client) GetActualMappingForIndexPattern(title string) ([]byte, error) {
+	if title == "" {
+		return nil, fmt.Errorf("index pattern title cannot be empty")
+	}
+
+	u := fmt.Sprintf("%s/api/index_patterns/_fields_for_wildcard?pattern=%s&meta_fields=_source&meta_fields=_id&meta_fields=_type&meta_fields=_index&meta_fields=_score", c.baseURL, title)
+	req, err := http.NewRequest("GET", u, nil)
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("Accept", "*/*")
+
+	resp, err := c.do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode >= 300 {
+		snippet, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
+		return nil, fmt.Errorf("kibana get index mapping failed with: %s — %s", resp.Status, strings.TrimSpace(string(snippet)))
+	}
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+	var fields IPFields
+	err = json.Unmarshal(body, &fields)
+	if err != nil {
+		return nil, err
+	}
+
+	fields_string, err := json.Marshal(fields.Fields)
+	if err != nil {
+		return nil, err
+	}
+	return fields_string, nil
+}
+
+func (c *Client) RefreshIndexPattern(id string, title string) error {
+	if id == "" {
+		return fmt.Errorf("index pattern id cannot be empty")
+	}
+	if title == "" {
+		return fmt.Errorf("index pattern title cannot be empty")
+	}
+	fields, err := c.GetActualMappingForIndexPattern(title)
+	if err != nil {
+		return err
+	}
+
+	if len(fields) == 0 {
+		return fmt.Errorf("no fields found for index pattern %s (index may not exist)", title)
+	}
+
+	u := fmt.Sprintf("%s/api/saved_objects/index-pattern/%s", c.baseURL, id)
+	body := map[string]any{
+		"attributes": map[string]any{
+			"title":         title,
+			"version":       utils.PatternVersion(),
+			"timeFieldName": "@timestamp",
+			"fields":        string(fields),
+		},
+	}
+	b, err := json.Marshal(body)
+	if err != nil {
+		return fmt.Errorf("failed to marshal index pattern body: %w", err)
+	}
+
+	req, err := http.NewRequest("PUT", u, bytes.NewReader(b))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	resp, err := c.do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode >= 300 {
+		snippet, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
+		return fmt.Errorf("kibana refresh index pattern failed: %s — %s", resp.Status, strings.TrimSpace(string(snippet)))
+	}
+	return nil
+}

pkg/opensearch/indices.go

@@ -52,7 +52,6 @@ func (c *Client) GetIndicesWithFields(pattern, fields string, sortBy ...string)
 	if sortParam != "" {
 		url += fmt.Sprintf("&s=%s", sortParam)
 	}
-
 	req, err := http.NewRequest("GET", url, nil)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create request: %v", err)