Add index correction internal logic

.github/workflows/dockers-index-job-correction.yml

@@ -0,0 +1,78 @@
+#
+# Copyright (C) 2019-2023 vdaas.org vald team <[email protected]>
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# You may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+name: "Build docker image: index-job-correction"
+on:
+  push:
+    branches:
+      - main
+    tags:
+      - "*.*.*"
+      - "v*.*.*"
+      - "*.*.*-*"
+      - "v*.*.*-*"
+    paths:
+      - ".github/actions/docker-build/actions.yaml"
+      - ".github/workflows/dockers-index-job-correction.yml"
+      - "go.mod"
+      - "go.sum"
+      - "internal/**"
+      - "!internal/**/*_test.go"
+      - "!internal/db/**"
+      - "!internal/k8s/**"
+      - "apis/grpc/**"
+      - "pkg/index/job/correction/**"
+      - "cmd/index/job/correction/**"
+      - "dockers/index/job/correction/Dockerfile"
+      - "versions/GO_VERSION"
+  pull_request:
+    paths:
+      - ".github/actions/docker-build/actions.yaml"
+      - ".github/workflows/_docker-image.yaml"
+      - ".github/workflows/dockers-index-job-correction.yml"
+      - "go.mod"
+      - "go.sum"
+      - "internal/**"
+      - "!internal/**/*_test.go"
+      - "!internal/db/**"
+      - "!internal/k8s/**"
+      - "apis/grpc/**"
+      - "pkg/index/job/correction/**"
+      - "cmd/index/job/correction/**"
+      - "dockers/index/job/correction/Dockerfile"
+      - "versions/GO_VERSION"
+  pull_request_target:
+    paths:
+      - ".github/actions/docker-build/actions.yaml"
+      - ".github/workflows/_docker-image.yaml"
+      - ".github/workflows/dockers-index-job-correction.yml"
+      - "go.mod"
+      - "go.sum"
+      - "internal/**"
+      - "!internal/**/*_test.go"
+      - "!internal/db/**"
+      - "!internal/k8s/**"
+      - "apis/grpc/**"
+      - "pkg/index/job/correction/**"
+      - "cmd/index/job/correction/**"
+      - "dockers/index/job/correction/Dockerfile"
+      - "versions/GO_VERSION"
+
+jobs:
+  build:
+    uses: ./.github/workflows/_docker-image.yaml
+    with:
+      target: index-job-correction
+    secrets: inherit

.golangci.yml

@@ -51,7 +51,6 @@ linters:
     - gochecknoinits
     - goconst
     - godot
-    - godox
     - gofumpt
     - goimports
     - gomnd
@@ -99,6 +98,7 @@ linters:
     # - gocognit
     # - gocritic
     # - gocyclo
+    # - godox
     # - goerr113
     # - gofmt
     # - goheader

Makefile

@@ -30,6 +30,7 @@ FILTER_GATEWAY_IMAGE            = $(NAME)-filter-gateway
 HELM_OPERATOR_IMAGE             = $(NAME)-helm-operator
 LB_GATEWAY_IMAGE                = $(NAME)-lb-gateway
 LOADTEST_IMAGE                  = $(NAME)-loadtest
+INDEX_JOB_CORRECTION_IMAGE      = $(NAME)-index-job-correction
 MANAGER_INDEX_IMAGE             = $(NAME)-manager-index
 MAINTAINER                      = "$(ORG).org $(NAME) team <$(NAME)@$(ORG).org>"
 

Makefile.d/build.mk

@@ -206,6 +206,35 @@ cmd/manager/index/index: \
 		$(dir $@)main.go
 	$@ -version
 
+cmd/index/job/correction/correction: \
+	$(GO_SOURCES_INTERNAL) \
+	$(PBGOS) \
+	$(shell find $(ROOTDIR)/cmd/index/job/correction/correction -type f -name '*.go' -not -name '*_test.go' -not -name 'doc.go') \
+	$(shell find $(ROOTDIR)/pkg/index/job/correction -type f -name '*.go' -not -name '*_test.go' -not -name 'doc.go')
+	$(eval CGO_ENABLED = 0)
+	CGO_ENABLED=$(CGO_ENABLED) \
+	GO111MODULE=on \
+	GOPRIVATE=$(GOPRIVATE) \
+	go build \
+		--ldflags "-w -extldflags=-static \
+		-X '$(GOPKG)/internal/info.Version=$(VERSION)' \
+		-X '$(GOPKG)/internal/info.GitCommit=$(GIT_COMMIT)' \
+		-X '$(GOPKG)/internal/info.BuildTime=$(DATETIME)' \
+		-X '$(GOPKG)/internal/info.GoVersion=$(GO_VERSION)' \
+		-X '$(GOPKG)/internal/info.GoOS=$(GOOS)' \
+		-X '$(GOPKG)/internal/info.GoArch=$(GOARCH)' \
+		-X '$(GOPKG)/internal/info.CGOEnabled=$(CGO_ENABLED)' \
+		-X '$(GOPKG)/internal/info.BuildCPUInfoFlags=$(CPU_INFO_FLAGS)' \
+		-buildid=" \
+		-mod=readonly \
+		-modcacherw \
+		-a \
+		-tags "osusergo netgo static_build" \
+		-trimpath \
+		-o $@ \
+		$(dir $@)main.go
+	$@ -version
+
 .PHONY: binary/build/zip
 ## build all binaries and zip them
 binary/build/zip: \

Makefile.d/docker.mk

@@ -188,3 +188,17 @@ docker/build/loadtest:
 	    -t $(ORG)/$(LOADTEST_IMAGE):$(TAG) . \
 	    --build-arg MAINTAINER=$(MAINTAINER) \
 	    --build-arg GO_VERSION=$(GO_VERSION)
+
+.PHONY: docker/name/index-job-correction
+docker/name/index-job-correction:
+	@echo "$(ORG)/$(INDEX_JOB_CORRECTION_IMAGE)"
+
+.PHONY: docker/build/index-job-correction
+## build index-job-correction image
+docker/build/index-job-correction:
+	$(DOCKER) build \
+	    $(DOCKER_OPTS) \
+	    -f dockers/index/job/correction/Dockerfile \
+	    -t $(ORG)/$(INDEX_JOB_CORRECTION_IMAGE):$(TAG) . \
+	    --build-arg MAINTAINER=$(MAINTAINER) \
+	    --build-arg GO_VERSION=$(GO_VERSION)

charts/vald/values.yaml

@@ -2631,3 +2631,38 @@ manager:
             net:
               dialer:
                 keepalive: 15m #indexer fetches uncommitted index length, which includes huge payload so we need to set keepalive longer than usual
+    # @schema {"name": "manager.index.corrector", "type": "object"}
+    corrector:
+      # @schema {"name": "manager.index.corrector.enabled", "type": "boolean"}
+      # manager.index.corrector.enabled -- enable index correction CronJob
+      enabled: false
+      # @schema {"name": "manager.index.corrector.check_duration", "type": "string"}
+      # manager.index.corrector.enabled -- check duration of index correction CronJob
+      check_duration: 24h
+      # @schema {"name": "manager.index.corrector.stream_list_concurrency", "type": "integer", "minimum": 1}
+      # manager.index.corrector.stream_list_concurrency -- concurrency for stream list object rpc
+      stream_list_concurrency: 200
+      # @schema {"name": "manager.index.corrector.bbolt_async_write_concurrency", "type": "integer", "minimum": 1}
+      # manager.index.corrector.bbolt_async_write_concurrency -- concurrency for bbolt async write
+      bbolt_async_write_concurrency: 2048
+      # @schema {"name": "manager.index.corrector.agent_namespace", "type": "string"}
+      # manager.index.corrector.agent_namespace -- namespace of agent pods to manage
+      agent_namespace: _MY_POD_NAMESPACE_
+      # @schema {"name": "manager.index.corrector.node_name", "type": "string"}
+      # manager.index.corrector.node_name -- node name
+      node_name: "" # _MY_NODE_NAME_
+      # @schema {"name": "manager.index.corrector.discoverer", "type": "object"}
+      discoverer:
+        # @schema {"name": "manager.index.corrector.discoverer.duration", "type": "string"}
+        # manager.index.corrector.discoverer.duration -- refresh duration to discover
+        duration: 500ms
+        # @schema {"name": "manager.index.corrector.discoverer.client", "alias": "grpc.client"}
+        # manager.index.corrector.discoverer.client -- gRPC client for discoverer (overrides defaults.grpc.client)
+        client: {}
+        # @schema {"name": "manager.index.corrector.discoverer.agent_client_options", "alias": "grpc.client"}
+        # manager.index.corrector.discoverer.agent_client_options -- gRPC client options for agents (overrides defaults.grpc.client)
+        agent_client_options:
+          dial_option:
+            net:
+              dialer:
+                keepalive: 15m #indexer fetches uncommitted index length, which includes huge payload so we need to set keepalive longer than usual

cmd/index/job/correction/main.go

@@ -0,0 +1,59 @@
+// Copyright (C) 2019-2023 vdaas.org vald team <[email protected]>
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// You may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package main
+
+import (
+	"context"
+
+	"github.com/vdaas/vald/internal/errors"
+	"github.com/vdaas/vald/internal/info"
+	"github.com/vdaas/vald/internal/log"
+	"github.com/vdaas/vald/internal/runner"
+	"github.com/vdaas/vald/internal/safety"
+	"github.com/vdaas/vald/pkg/index/job/correction/config"
+	"github.com/vdaas/vald/pkg/index/job/correction/usecase"
+)
+
+const (
+	maxVersion = "v0.0.10"
+	minVersion = "v0.0.0"
+	name       = "index correction job"
+)
+
+func main() {
+	if err := safety.RecoverFunc(func() error {
+		return runner.Do(
+			context.Background(),
+			runner.WithName(name),
+			runner.WithVersion(info.Version, maxVersion, minVersion),
+			runner.WithConfigLoader(func(path string) (interface{}, *config.GlobalConfig, error) {
+				cfg, err := config.NewConfig(path)
+				if err != nil {
+					return nil, nil, errors.Wrap(err, "failed to load "+name+"'s configuration")
+				}
+				return cfg, &cfg.GlobalConfig, nil
+			}),
+			runner.WithDaemonInitializer(func(cfg interface{}) (runner.Runner, error) {
+				c, ok := cfg.(*config.Data)
+				if !ok {
+					return nil, errors.ErrInvalidConfig
+				}
+				return usecase.New(c)
+			}),
+		)
+	})(); err != nil {
+		log.Fatal(err, info.Get())
+		return
+	}
+}