Add index name limit

quesma/elasticsearch/index.go

@@ -3,6 +3,8 @@
 package elasticsearch
 
 import (
+	"fmt"
+	"quesma/end_user_errors"
 	"strings"
 )
 
@@ -20,3 +22,15 @@ func IsInternalIndex(index string) bool {
 
 // InternalPaths is a list of paths that are considered internal and should not handled by Quesma
 var InternalPaths = []string{"/_nodes", "/_xpack"}
+
+func IsValidIndexName(name string) error {
+	const maxIndexNameLength = 256
+
+	if len(name) > maxIndexNameLength {
+		return end_user_errors.ErrIndexNameTooLong.New(fmt.Errorf("index name is too long: %d, max length: %d", len(name), maxIndexNameLength))
+	}
+
+	// TODO add more checks, elasticsearch is quite strict about index names
+
+	return nil
+}

quesma/elasticsearch/index_test.go

@@ -0,0 +1,35 @@
+// Copyright Quesma, licensed under the Elastic License 2.0.
+// SPDX-License-Identifier: Elastic-2.0
+package elasticsearch
+
+import "testing"
+
+func TestIsValidIndexName(t *testing.T) {
+
+	tests := []struct {
+		name    string
+		wantErr bool
+	}{
+		{
+			name:    "foo",
+			wantErr: false,
+		},
+		{
+			name:    "foo_bar",
+			wantErr: false,
+		},
+		{
+			name:    "esc_base_agent_client_cloud_container_data_stream_destination_device_dll_dns_ecs_email_error_event_faas_file_group_host_http_log_network_observer_orchestrator_organization_package_process_registry_related_rule_server_service_source_threat_tls_url_user_user_agent_volume_vulnerability_windows",
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if err := IsValidIndexName(tt.name); (err != nil) != tt.wantErr {
+				t.Errorf("IsValidIndexName() error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+
+}

quesma/end_user_errors/clickhouse.go

@@ -48,6 +48,10 @@ func GuessClickhouseErrorType(err error) *EndUserError {
 			return ErrDatabaseTLSVerify.New(originalErr)
 		}
 
+		if strings.Contains(s, "code: 76") {
+			return ErrDatabaseStorageError.New(originalErr)
+		}
+
 		err = errors.Unwrap(err)
 		if err == nil {
 			break

quesma/end_user_errors/end_user.go

@@ -96,6 +96,7 @@ var ErrNoSuchTable = errorType(2002, "Missing table.")
 var ErrNoSuchSchema = errorType(2003, "Missing schema.")
 var ErrNoIngest = errorType(2004, "Ingest is not enabled.")
 var ErrNoConnector = errorType(2005, "No connector found.")
+var ErrIndexNameTooLong = errorType(2006, "Index name is too long.")
 
 var ErrDatabaseTableNotFound = errorType(3001, "Table not found in database.")
 var ErrDatabaseFieldNotFound = errorType(3002, "Field not found in database.")
@@ -106,3 +107,4 @@ var ErrDatabaseOtherError = errorType(3006, "Database query has failed. You may
 var ErrDatabaseInvalidProtocol = errorType(3007, "Invalid database protocol. Check your connection settings. ")
 var ErrDatabaseTLS = errorType(3008, "Error establishing TLS connection with database. Check your connection settings.")
 var ErrDatabaseTLSVerify = errorType(3009, "Error verifying TLS certificate with database. Check your connection settings.")
+var ErrDatabaseStorageError = errorType(3010, "Error storing data in database. Check your database settings.")

quesma/ingest/processor.go

@@ -11,6 +11,7 @@ import (
 	chLib "quesma/clickhouse"
 	"quesma/comment_metadata"
 	"quesma/common_table"
+	"quesma/elasticsearch"
 	"quesma/end_user_errors"
 	"quesma/jsonprocessor"
 	"quesma/logger"
@@ -688,6 +689,11 @@ func (ip *IngestProcessor) processInsertQuery(ctx context.Context,
 
 func (lm *IngestProcessor) Ingest(ctx context.Context, indexName string, jsonData []types.JSON) error {
 
+	err := elasticsearch.IsValidIndexName(indexName)
+	if err != nil {
+		return err
+	}
+
 	nameFormatter := DefaultColumnNameFormatter()
 	transformer := jsonprocessor.IngestTransformerFor(indexName, lm.cfg)
 	return lm.ProcessInsertQuery(ctx, indexName, jsonData, transformer, nameFormatter)

quesma/quesma/router.go

@@ -283,6 +283,11 @@ func ConfigureRouter(cfg *config.QuesmaConfiguration, sr schema.Registry, lm *cl
 		case "PUT":
 			index := req.Params["index"]
 
+			err := elasticsearch.IsValidIndexName(index)
+			if err != nil {
+				return nil, err
+			}
+
 			body, err := types.ExpectJSON(req.ParsedBody)
 			if err != nil {
 				return nil, err
@@ -401,6 +406,12 @@ func ConfigureRouter(cfg *config.QuesmaConfiguration, sr schema.Registry, lm *cl
 		case "PUT":
 
 			index := req.Params["index"]
+
+			err := elasticsearch.IsValidIndexName(index)
+			if err != nil {
+				return nil, err
+			}
+
 			if req.Body == "" {
 				logger.Warn().Msgf("empty body in PUT /%s request, Quesma is not doing anything", index)
 				return putIndexResult(index)

quesma/quesma/router_v2.go

@@ -293,6 +293,11 @@ func ConfigureSearchRouterV2(cfg *config.QuesmaConfiguration, dependencies quesm
 		case "PUT":
 			index := req.Params["index"]
 
+			err := elasticsearch.IsValidIndexName(index)
+			if err != nil {
+				return nil, err
+			}
+
 			body, err := types.ExpectJSON(req.ParsedBody)
 			if err != nil {
 				return nil, err
@@ -416,6 +421,11 @@ func ConfigureSearchRouterV2(cfg *config.QuesmaConfiguration, dependencies quesm
 				return putIndexResult(index)
 			}
 
+			err := elasticsearch.IsValidIndexName(index)
+			if err != nil {
+				return nil, err
+			}
+
 			body, err := types.ExpectJSON(req.ParsedBody)
 			if err != nil {
 				return nil, err