opencloud/services/search/pkg/command/server.go

package command

import (
	"context"
	"crypto/tls"
	"fmt"
	"net/http"
	"os"
	"os/signal"
	"time"

	"github.com/opencloud-eu/opencloud/pkg/config/configlog"
	"github.com/opencloud-eu/opencloud/pkg/generators"
	"github.com/opencloud-eu/opencloud/pkg/log"
	"github.com/opencloud-eu/opencloud/pkg/registry"
	"github.com/opencloud-eu/opencloud/pkg/runner"
	ogrpc "github.com/opencloud-eu/opencloud/pkg/service/grpc"
	"github.com/opencloud-eu/opencloud/pkg/tracing"
	"github.com/opencloud-eu/opencloud/pkg/version"
	"github.com/opencloud-eu/opencloud/services/search/pkg/bleve"
	"github.com/opencloud-eu/opencloud/services/search/pkg/config"
	"github.com/opencloud-eu/opencloud/services/search/pkg/config/parser"
	"github.com/opencloud-eu/opencloud/services/search/pkg/content"
	searchmapping "github.com/opencloud-eu/opencloud/services/search/pkg/mapping"
	"github.com/opencloud-eu/opencloud/services/search/pkg/metrics"
	"github.com/opencloud-eu/opencloud/services/search/pkg/opensearch"
	bleveQuery "github.com/opencloud-eu/opencloud/services/search/pkg/query/bleve"
	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
	"github.com/opencloud-eu/opencloud/services/search/pkg/server/debug"
	"github.com/opencloud-eu/opencloud/services/search/pkg/server/grpc"
	svcEvent "github.com/opencloud-eu/opencloud/services/search/pkg/service/event"

	"github.com/opencloud-eu/reva/v2/pkg/events/raw"
	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
	opensearchgo "github.com/opensearch-project/opensearch-go/v4"
	opensearchgoAPI "github.com/opensearch-project/opensearch-go/v4/opensearchapi"
	"github.com/spf13/cobra"
)

// Server is the entrypoint for the server command.
func Server(cfg *config.Config) *cobra.Command {
	return &cobra.Command{
		Use:   "server",
		Short: fmt.Sprintf("start the %s service without runtime (unsupervised mode)", cfg.Service.Name),
		PreRunE: func(cmd *cobra.Command, args []string) error {
			return configlog.ReturnFatal(parser.ParseConfig(cfg))
		},
		RunE: func(cmd *cobra.Command, args []string) error {
			logger := log.Configure(cfg.Service.Name, cfg.Commons, cfg.LogLevel)
			traceProvider, err := tracing.GetTraceProvider(cmd.Context(), cfg.Commons.TracesExporter, cfg.Service.Name)
			if err != nil {
				return err
			}

			cfg.GrpcClient, err = ogrpc.NewClient(
				append(ogrpc.GetClientOptions(cfg.GRPCClientTLS), ogrpc.WithTraceProvider(traceProvider))...,
			)
			if err != nil {
				return err
			}

			var cancel context.CancelFunc
			if cfg.Context == nil {
				cfg.Context, cancel = signal.NotifyContext(context.Background(), runner.StopSignals...)
				defer cancel()
			}
			ctx := cfg.Context

			mtrcs := metrics.New()
			mtrcs.BuildInfo.WithLabelValues(version.GetString()).Set(1)

			// initialize search engine
			var eng search.Engine
			switch cfg.Engine.Type {
			case "bleve":
				idx, classification, err := bleve.NewIndex(cfg.Engine.Bleve.Datapath)
				// warn before the error check: the new mapping may already be
				// persisted, then later startups classify equal and stay silent
				if classification.Verdict == searchmapping.VerdictAdditive {
					logger.Warn().
						Strs("fields", classification.NewFields).
						Msgf("the bleve index at %s was built with an older schema; the new fields were added to the index schema, but documents indexed before the upgrade do not contain them and queries on these fields will miss those documents until they are re-indexed; to re-index everything run: opencloud search index --all-spaces --force-rescan", cfg.Engine.Bleve.Datapath)
				}
				if err != nil {
					return err
				}

				defer func() {
					if err = idx.Close(); err != nil {
						logger.Error().Err(err).Msg("could not close bleve index")
					}
				}()

				eng = bleve.NewBackend(idx, bleveQuery.DefaultCreator, logger)
			case "open-search":
				clientConfig := opensearchgo.Config{
					Addresses:             cfg.Engine.OpenSearch.Client.Addresses,
					Username:              cfg.Engine.OpenSearch.Client.Username,
					Password:              cfg.Engine.OpenSearch.Client.Password,
					Header:                cfg.Engine.OpenSearch.Client.Header,
					RetryOnStatus:         cfg.Engine.OpenSearch.Client.RetryOnStatus,
					DisableRetry:          cfg.Engine.OpenSearch.Client.DisableRetry,
					EnableRetryOnTimeout:  cfg.Engine.OpenSearch.Client.EnableRetryOnTimeout,
					MaxRetries:            cfg.Engine.OpenSearch.Client.MaxRetries,
					CompressRequestBody:   cfg.Engine.OpenSearch.Client.CompressRequestBody,
					DiscoverNodesOnStart:  cfg.Engine.OpenSearch.Client.DiscoverNodesOnStart,
					DiscoverNodesInterval: cfg.Engine.OpenSearch.Client.DiscoverNodesInterval,
					EnableMetrics:         cfg.Engine.OpenSearch.Client.EnableMetrics,
					EnableDebugLogger:     cfg.Engine.OpenSearch.Client.EnableDebugLogger,
					Transport: &http.Transport{
						TLSClientConfig: &tls.Config{
							MinVersion:         tls.VersionTLS12,
							InsecureSkipVerify: cfg.Engine.OpenSearch.Client.Insecure,
						},
					},
				}

				if cfg.Engine.OpenSearch.Client.CACert != "" {
					certBytes, err := os.ReadFile(cfg.Engine.OpenSearch.Client.CACert)
					if err != nil {
						return fmt.Errorf("failed to read CA cert: %w", err)
					}
					clientConfig.CACert = certBytes
				}

				client, err := opensearchgoAPI.NewClient(opensearchgoAPI.Config{Client: clientConfig})
				if err != nil {
					return fmt.Errorf("failed to create OpenSearch client: %w", err)
				}

				// a hung cluster must fail the start, not block it forever
				startupCtx, cancelStartup := context.WithTimeout(ctx, time.Minute)
				indexName := opensearch.VersionedIndexName(cfg.Engine.OpenSearch.ResourceIndex.Name)
				openSearchBackend, err := opensearch.NewBackend(startupCtx, indexName, client, logger)
				cancelStartup()
				if err != nil {
					return fmt.Errorf("failed to create OpenSearch backend: %w", err)
				}

				eng = openSearchBackend
			default:
				return fmt.Errorf("unknown search engine: %s", cfg.Engine.Type)
			}

			// initialize gateway selector
			selector, err := pool.GatewaySelector(cfg.Reva.Address, pool.WithRegistry(registry.GetRegistry()), pool.WithTracerProvider(traceProvider))
			if err != nil {
				logger.Fatal().Err(err).Msg("could not get reva gateway selector")
				return err
			}

			// initialize search content extractor
			var extractor content.Extractor
			switch cfg.Extractor.Type {
			case "basic":
				if extractor, err = content.NewBasicExtractor(logger); err != nil {
					return err
				}
			case "tika":
				if extractor, err = content.NewTikaExtractor(selector, logger, cfg); err != nil {
					return err
				}
			default:
				return fmt.Errorf("unknown search extractor: %s", cfg.Extractor.Type)
			}

			ss := search.NewService(selector, eng, extractor, mtrcs, logger, cfg)

			// setup the servers
			gr := runner.NewGroup()

			if !cfg.GRPC.Disabled {
				grpcServer, err := grpc.Server(
					grpc.Config(cfg),
					grpc.Logger(logger),
					grpc.Name(cfg.Service.Name),
					grpc.Context(ctx),
					grpc.Metrics(mtrcs),
					grpc.JWTSecret(cfg.TokenManager.JWTSecret),
					grpc.TraceProvider(traceProvider),
					grpc.GatewaySelector(selector),
					grpc.Searcher(ss),
				)
				if err != nil {
					logger.Error().Err(err).Str("transport", "grpc").Msg("Failed to initialize server")
					return err
				}

				gr.Add(runner.NewGoMicroGrpcServerRunner(cfg.Service.Name+".grpc", grpcServer))
			} else {
				logger.Info().Msg("gRPC server disabled, not starting gRPC service")
			}

			if !cfg.Events.Disabled {
				connName := generators.GenerateConnectionName(cfg.Service.Name, generators.NTypeBus)
				bus, err := raw.FromConfig(context.Background(), connName, raw.Config{
					Endpoint:             cfg.Events.Endpoint,
					Cluster:              cfg.Events.Cluster,
					EnableTLS:            cfg.Events.EnableTLS,
					TLSInsecure:          cfg.Events.TLSInsecure,
					TLSRootCACertificate: cfg.Events.TLSRootCACertificate,
					AuthUsername:         cfg.Events.AuthUsername,
					AuthPassword:         cfg.Events.AuthPassword,
					MaxAckPending:        cfg.Events.MaxAckPending,
					AckWait:              cfg.Events.AckWait,
				})
				if err != nil {
					logger.Error().Err(err).Msg("Failed to create event bus client")
					return err
				}

				eventSvc, err := svcEvent.New(ctx, bus, logger, traceProvider, mtrcs, ss, cfg.Events.DebounceDuration, cfg.Events.NumConsumers, cfg.Events.AsyncUploads)
				if err != nil {
					logger.Error().Err(err).Str("transport", "event").Msg("Failed to initialize server")
					return err
				}

				gr.Add(runner.New(cfg.Service.Name+".svc", func() error {
					return eventSvc.Run()
				}, func() {
					eventSvc.Close()
				}))
			} else {
				logger.Info().Msg("event listening disabled, not starting event service")
			}

			// always start a debug server
			{
				debugServer, err := debug.Server(
					debug.Logger(logger),
					debug.Context(ctx),
					debug.Config(cfg),
				)
				if err != nil {
					logger.Error().Err(err).Str("transport", "debug").Msg("Failed to initialize server")
					return err
				}

				gr.Add(runner.NewGolangHttpServerRunner(cfg.Service.Name+".debug", debugServer))
			}

			grResults := gr.Run(ctx)

			// return the first non-nil error found in the results
			for _, grResult := range grResults {
				if grResult.RunnerError != nil {
					return grResult.RunnerError
				}
			}
			return nil
		},
	}
}