Reindex disabled spaces once they get enabled again

protogen/gen/opencloud/services/search/v0/search.swagger.json

@@ -46,7 +46,7 @@
                   "$ref": "#/definitions/v0IndexSpaceResponse"
                 },
                 "error": {
-                  "$ref": "#/definitions/rpcStatus"
+                  "$ref": "#/definitions/googlerpcStatus"
                 }
               },
               "title": "Stream result of v0IndexSpaceResponse"
@@ -55,7 +55,7 @@
           "default": {
             "description": "An unexpected error response.",
             "schema": {
-              "$ref": "#/definitions/rpcStatus"
+              "$ref": "#/definitions/googlerpcStatus"
             }
           }
         },
@@ -87,7 +87,7 @@
           "default": {
             "description": "An unexpected error response.",
             "schema": {
-              "$ref": "#/definitions/rpcStatus"
+              "$ref": "#/definitions/googlerpcStatus"
             }
           }
         },
@@ -119,7 +119,7 @@
           "default": {
             "description": "An unexpected error response.",
             "schema": {
-              "$ref": "#/definitions/rpcStatus"
+              "$ref": "#/definitions/googlerpcStatus"
             }
           }
         },
@@ -140,16 +140,7 @@
     }
   },
   "definitions": {
-    "protobufAny": {
-      "type": "object",
-      "properties": {
-        "@type": {
-          "type": "string"
-        }
-      },
-      "additionalProperties": {}
-    },
-    "rpcStatus": {
+    "googlerpcStatus": {
       "type": "object",
       "properties": {
         "code": {
@@ -167,6 +158,15 @@
         }
       }
     },
+    "protobufAny": {
+      "type": "object",
+      "properties": {
+        "@type": {
+          "type": "string"
+        }
+      },
+      "additionalProperties": {}
+    },
     "v0Audio": {
       "type": "object",
       "properties": {
@@ -382,9 +382,24 @@
         "error": {
           "type": "string",
           "description": "Contains an error message in case indexing this particular space failed."
+        },
+        "status": {
+          "$ref": "#/definitions/v0IndexSpaceResponseStatus",
+          "description": "The outcome of indexing this particular space."
         }
       }
     },
+    "v0IndexSpaceResponseStatus": {
+      "type": "string",
+      "enum": [
+        "STATUS_UNKNOWN",
+        "STATUS_SUCCESS",
+        "STATUS_SKIPPED",
+        "STATUS_ERROR"
+      ],
+      "default": "STATUS_UNKNOWN",
+      "description": " - STATUS_SUCCESS: The space has been indexed successfully.\n - STATUS_SKIPPED: The space has been skipped because it is disabled. It will be indexed\nonce it gets enabled again.\n - STATUS_ERROR: Indexing the space failed, see the error field for details."
+    },
     "v0LivePhoto": {
       "type": "object",
       "properties": {

protogen/proto/opencloud/services/search/v0/search.proto

@@ -112,6 +112,17 @@ message IndexSpaceRequest {
 }
 
 message IndexSpaceResponse {
+  enum Status {
+    STATUS_UNKNOWN = 0;
+    // The space has been indexed successfully.
+    STATUS_SUCCESS = 1;
+    // The space has been skipped because it is disabled. It will be indexed
+    // once it gets enabled again.
+    STATUS_SKIPPED = 2;
+    // Indexing the space failed, see the error field for details.
+    STATUS_ERROR = 3;
+  }
+
   // The id of the space that has just been indexed.
   string space_id = 1;
   // The duration it took to index this space.
@@ -122,4 +133,6 @@ message IndexSpaceResponse {
   int64 total_spaces = 4;
   // Contains an error message in case indexing this particular space failed.
   string error = 5;
+  // The outcome of indexing this particular space.
+  Status status = 6;
 }

services/search/pkg/command/index.go

@@ -8,6 +8,7 @@ import (
 	"io"
 	"os"
 	"os/signal"
+	"strconv"
 	"syscall"
 
 	"github.com/opencloud-eu/opencloud/pkg/config/configlog"
@@ -92,14 +93,7 @@ func Index(cfg *config.Config) *cobra.Command {
 					return err
 				}
 
-				if progress.GetError() != "" {
-					fmt.Printf("[%d/%d] failed to index space %s: %s\n",
-						progress.GetIndexedSpaces(), progress.GetTotalSpaces(), progress.GetSpaceId(), progress.GetError())
-					continue
-				}
-
-				fmt.Printf("[%d/%d] indexed space %s in %s\n",
-					progress.GetIndexedSpaces(), progress.GetTotalSpaces(), progress.GetSpaceId(), progress.GetSpaceDuration().AsDuration())
+				printProgress(progress)
 			}
 			return nil
 		},
@@ -138,3 +132,30 @@ func Index(cfg *config.Config) *cobra.Command {
 
 	return indexCmd
 }
+
+// printProgress prints a single progress line, e.g.
+//
+//	[ 1/12 SKIPPED] <space id> is disabled, it will be indexed once it is enabled again
+//	[ 2/12 SUCCESS] <space id> indexed in 40.9ms
+//	[ 3/12 ERROR  ] <space id> failed: <error>
+//
+// The counter is padded to the width of the total and the status to the
+// longest status so that the messages line up.
+func printProgress(progress *searchsvc.IndexSpaceResponse) {
+	var status, msg string
+	switch {
+	case progress.GetStatus() == searchsvc.IndexSpaceResponse_STATUS_SKIPPED:
+		status = "SKIPPED"
+		msg = progress.GetSpaceId() + " is disabled, it will be indexed once it is enabled again"
+	// servers not setting a status only report failures via the error field
+	case progress.GetStatus() == searchsvc.IndexSpaceResponse_STATUS_ERROR || progress.GetError() != "":
+		status = "ERROR"
+		msg = progress.GetSpaceId() + " failed: " + progress.GetError()
+	default:
+		status = "SUCCESS"
+		msg = fmt.Sprintf("%s indexed in %s", progress.GetSpaceId(), progress.GetSpaceDuration().AsDuration())
+	}
+
+	width := len(strconv.FormatInt(progress.GetTotalSpaces(), 10))
+	fmt.Printf("[%*d/%d %-7s] %s\n", width, progress.GetIndexedSpaces(), progress.GetTotalSpaces(), status, msg)
+}

services/search/pkg/command/server.go

@@ -29,6 +29,8 @@ import (
 	"github.com/opencloud-eu/reva/v2/pkg/events/raw"
 	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
 	"github.com/spf13/cobra"
+
+	"github.com/nats-io/nats.go/jetstream"
 )
 
 // Server is the entrypoint for the server command.
@@ -122,6 +124,40 @@ func Server(cfg *config.Config) *cobra.Command {
 
 			ss := search.NewService(selector, eng, extractor, mtrcs, logger, cfg)
 
+			// The KV bucket is used to remember spaces that were skipped during
+			// (re)indexing because they were disabled. It is shared between the gRPC
+			// reindex handler (which fills it) and the event consumer (which drains it
+			// again once a space gets enabled).
+			rawEventsCfg := 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,
+			}
+			skippedSpaces := search.NewSkippedSpaces(nil)
+			if !cfg.Events.Disabled {
+				kvConnName := generators.GenerateConnectionName(cfg.Service.Name, generators.NTypeKeyValue)
+				js, err := raw.JetStream(ctx, kvConnName, rawEventsCfg)
+				if err != nil {
+					logger.Error().Err(err).Msg("Failed to connect to NATS jetstream")
+					return err
+				}
+				kv, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
+					Bucket:      search.SkippedSpacesBucket,
+					Description: "Spaces that were skipped during (re)indexing because they were disabled",
+				})
+				if err != nil {
+					logger.Error().Err(err).Msg("Failed to create the skipped spaces KV bucket")
+					return err
+				}
+				skippedSpaces = search.NewSkippedSpaces(kv)
+			}
+
 			// setup the servers
 			gr := runner.NewGroup()
 
@@ -136,6 +172,7 @@ func Server(cfg *config.Config) *cobra.Command {
 					grpc.TraceProvider(traceProvider),
 					grpc.GatewaySelector(selector),
 					grpc.Searcher(ss),
+					grpc.SkippedSpaces(skippedSpaces),
 				)
 				if err != nil {
 					logger.Error().Err(err).Str("transport", "grpc").Msg("Failed to initialize server")
@@ -149,23 +186,13 @@ func Server(cfg *config.Config) *cobra.Command {
 
 			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,
-				})
+				bus, err := raw.FromConfig(context.Background(), connName, rawEventsCfg)
 				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)
+				eventSvc, err := svcEvent.New(ctx, bus, logger, traceProvider, mtrcs, ss, skippedSpaces, 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

services/search/pkg/search/skippedspaces.go

@@ -0,0 +1,136 @@
+package search
+
+import (
+	"context"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"time"
+
+	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/opencloud-eu/reva/v2/pkg/storagespace"
+)
+
+const (
+	SkippedSpacesBucket  = "search-skipped-spaces"
+	skippedSpacesTimeout = 10 * time.Second
+)
+
+// SkippedSpaces keeps track of spaces that were skipped during (re)indexing
+// because they were disabled. When such a space gets enabled again it can be
+// looked up here and reindexed.
+type SkippedSpaces struct {
+	kv jetstream.KeyValue
+}
+
+// skippedSpace is the value stored in the KV bucket for a skipped space.
+type skippedSpace struct {
+	SpaceID      string `json:"spaceId"`
+	ForceReindex bool   `json:"forceReindex"`
+}
+
+// NewSkippedSpaces creates a new SkippedSpaces tracker backed by the given
+// NATS JS KV bucket. A nil bucket turns all operations into no-ops, which is
+// handy for tests and for deployments that run without an event bus.
+func NewSkippedSpaces(kv jetstream.KeyValue) *SkippedSpaces {
+	return &SkippedSpaces{kv: kv}
+}
+
+// key returns the canonical, KV-safe key for the given space id. NATS KV keys
+// may only contain a limited set of characters, so the (possibly `$`/`!`
+// separated) space id is normalized and base64url encoded.
+func (s *SkippedSpaces) key(spaceID *provider.StorageSpaceId) (string, error) {
+	rid, err := storagespace.ParseID(spaceID.GetOpaqueId())
+	if err != nil {
+		return "", err
+	}
+	canonical := storagespace.FormatResourceID(&provider.ResourceId{
+		StorageId: rid.GetStorageId(),
+		SpaceId:   rid.GetSpaceId(),
+	})
+	return base64.RawURLEncoding.EncodeToString([]byte(canonical)), nil
+}
+
+// Mark records that the given space was skipped while it was disabled.
+// forceReindex indicates whether the skipped scan was a forced reindex. An
+// existing forced mark is never downgraded to a shallow one.
+func (s *SkippedSpaces) Mark(spaceID *provider.StorageSpaceId, forceReindex bool) error {
+	if s == nil || s.kv == nil {
+		return nil
+	}
+	key, err := s.key(spaceID)
+	if err != nil {
+		return err
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), skippedSpacesTimeout)
+	defer cancel()
+
+	if !forceReindex {
+		entry, err := s.kv.Get(ctx, key)
+		switch {
+		case err == nil:
+			if decodeSkippedSpace(entry.Value()).ForceReindex {
+				return nil
+			}
+		case !errors.Is(err, jetstream.ErrKeyNotFound):
+			return err
+		}
+	}
+
+	value, err := json.Marshal(skippedSpace{SpaceID: spaceID.GetOpaqueId(), ForceReindex: forceReindex})
+	if err != nil {
+		return err
+	}
+	_, err = s.kv.Put(ctx, key, value)
+	return err
+}
+
+// decodeSkippedSpace decodes a KV value. Values that can not be decoded are
+// treated as shallow (non-forced) marks.
+func decodeSkippedSpace(value []byte) skippedSpace {
+	var ss skippedSpace
+	_ = json.Unmarshal(value, &ss)
+	return ss
+}
+
+// Unmark removes any record for the given space. It is a no-op if the space
+// was not tracked.
+func (s *SkippedSpaces) Unmark(spaceID *provider.StorageSpaceId) error {
+	if s == nil || s.kv == nil {
+		return nil
+	}
+	key, err := s.key(spaceID)
+	if err != nil {
+		return err
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), skippedSpacesTimeout)
+	defer cancel()
+	if err := s.kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
+		return err
+	}
+	return nil
+}
+
+// IsMarked reports whether the given space was previously skipped while it was
+// disabled, and whether the skipped scan was a forced reindex.
+func (s *SkippedSpaces) IsMarked(spaceID *provider.StorageSpaceId) (marked bool, forceReindex bool, err error) {
+	if s == nil || s.kv == nil {
+		return false, false, nil
+	}
+	key, err := s.key(spaceID)
+	if err != nil {
+		return false, false, err
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), skippedSpacesTimeout)
+	defer cancel()
+	entry, err := s.kv.Get(ctx, key)
+	switch {
+	case err == nil:
+		return true, decodeSkippedSpace(entry.Value()).ForceReindex, nil
+	case errors.Is(err, jetstream.ErrKeyNotFound):
+		return false, false, nil
+	default:
+		return false, false, err
+	}
+}