Merge branch 'main' into feature/benchmark/add-benchmark-core-ngt

.github/ISSUE_TEMPLATE/bug_report.md

@@ -22,7 +22,7 @@ assignees: ""
 
 <!--- Please change the versions below along with your environment -->
 
-- Go Version: 1.21.3
+- Go Version: 1.21.4
 - Docker Version: 20.10.8
-- Kubernetes Version: v1.28.2
-- NGT Version: 2.1.3
+- Kubernetes Version: v1.28.4
+- NGT Version: 2.1.5

.github/ISSUE_TEMPLATE/security_issue_report.md

@@ -16,7 +16,7 @@ assignees: ""
 
 <!--- Please change the versions below along with your environment -->
 
-- Go Version: 1.21.3
+- Go Version: 1.21.4
 - Docker Version: 20.10.8
-- Kubernetes Version: v1.28.2
-- NGT Version: 2.1.3
+- Kubernetes Version: v1.28.4
+- NGT Version: 2.1.5

.github/PULL_REQUEST_TEMPLATE.md

@@ -16,10 +16,10 @@
 
 <!--- Please change the versions below along with your environment -->
 
-- Go Version: 1.21.3
+- Go Version: 1.21.4
 - Docker Version: 20.10.8
-- Kubernetes Version: v1.28.2
-- NGT Version: 2.1.3
+- Kubernetes Version: v1.28.4
+- NGT Version: 2.1.5
 
 ### Checklist:
 

.golangci.yml

@@ -42,7 +42,6 @@ linters:
     - errorlint
     - execinquery
     - exhaustive
-    - exhaustruct
     - exportloopref
     - forcetypeassert
     - ginkgolinter
@@ -92,6 +91,7 @@ linters:
     # Disabled by your configuration linters
     # - cyclop
     # - errcheck
+    # - exhaustruct
     # - forbidigo
     # - funlen
     # - gci

Makefile.d/functions.mk

@@ -37,12 +37,12 @@ define proto-code-gen
 		--plugin protoc-gen-go="$(GOBIN)/protoc-gen-go" \
                 --go-vtproto_out=$(GOPATH)/src \
 		--plugin protoc-gen-go-vtproto="$(GOBIN)/protoc-gen-go-vtproto" \
-                --go-vtproto_opt=features=grpc+marshal+unmarshal+size+equal+clone+pool \
-                --go-vtproto_opt=pool=$(ROOTDIR)/apis/proto/v1/payload.Object.Vector \
+                --go-vtproto_opt=features=grpc+marshal+unmarshal+size+equal+clone \
 		$1
 endef
 		# --go-grpc_out=$(GOPATH)/src \
 		# --plugin protoc-gen-go-grpc="${GOBIN}/protoc-gen-go-grpc" \
+                # --go-vtproto_opt=pool=$(ROOTDIR)/apis/proto/v1/payload.Object.Vector \
                 # --go-vtproto_opt=pool=$(ROOTDIR)/apis/proto/v1/payload.Insert.MultiRequest \
                 # --go-vtproto_opt=pool=$(ROOTDIR)/apis/proto/v1/payload.Insert.Request \
                 # --go-vtproto_opt=pool=$(ROOTDIR)/apis/proto/v1/payload.Object.Vector \

Makefile.d/k8s.mk

@@ -43,6 +43,8 @@ k8s/manifest/update: \
 	mv $(TEMP_DIR)/vald/templates/gateway/lb k8s/gateway/lb
 	mv $(TEMP_DIR)/vald/templates/manager/index k8s/manager/index
 	mv $(TEMP_DIR)/vald/templates/index/job/correction k8s/index/job/correction
+	mv $(TEMP_DIR)/vald/templates/index/job/creation k8s/index/job/creation
+	mv $(TEMP_DIR)/vald/templates/index/job/save k8s/index/job/save
 	rm -rf $(TEMP_DIR)
 
 .PHONY: k8s/manifest/helm-operator/clean

Makefile.d/proto.mk

@@ -166,8 +166,6 @@ $(PBGOS): \
 	find $(ROOTDIR)/apis/grpc/* -name '*.go' | xargs sed -i -E "s%google.golang.org/grpc/status%github.com/vdaas/vald/internal/net/grpc/status%g"
 	find $(ROOTDIR)/apis/grpc/* -name '*.go' | xargs sed -i -E "s%\"io\"%\"github.com/vdaas/vald/internal/io\"%g"
 	find $(ROOTDIR)/apis/grpc/* -name '*.go' | xargs sed -i -E "s%\"sync\"%\"github.com/vdaas/vald/internal/sync\"%g"
-	find $(ROOTDIR)/apis/grpc/* -name '*.go' | xargs sed -i -E "s/Vector = &Object_Vector\{\}/Vector = Object_VectorFromVTPool\(\)/g"
-	find $(ROOTDIR)/apis/grpc/* -name '*.go' | xargs sed -i -E "s/v := &Object_Vector\{\}/v := Object_VectorFromVTPool\(\)/g"
 
 $(SWAGGERS): \
 	$(PROTOS) \

apis/grpc/v1/agent/core/agent.go

@@ -22,4 +22,5 @@ const (
 	SaveIndexRPCName          = "SaveIndex"
 	CreateAndSaveIndexRPCName = "CreateAndSaveIndex"
 	IndexInfoRPCName          = "IndexInfo"
+	GetTimestampRPCName       = "GetTimestamp"
 )

apis/grpc/v1/agent/core/agent_vtproto.pb.go

@@ -50,6 +50,8 @@ type AgentClient interface {
 	CreateAndSaveIndex(ctx context.Context, in *payload.Control_CreateIndexRequest, opts ...grpc.CallOption) (*payload.Empty, error)
 	// Represent the RPC to get the agent index information.
 	IndexInfo(ctx context.Context, in *payload.Empty, opts ...grpc.CallOption) (*payload.Info_Index_Count, error)
+	// Represent the RPC to get the vector metadata. This RPC is mainly used for index correction process
+	GetTimestamp(ctx context.Context, in *payload.Object_GetTimestampRequest, opts ...grpc.CallOption) (*payload.Object_Timestamp, error)
 }
 
 type agentClient struct {
@@ -96,6 +98,15 @@ func (c *agentClient) IndexInfo(ctx context.Context, in *payload.Empty, opts ...
 	return out, nil
 }
 
+func (c *agentClient) GetTimestamp(ctx context.Context, in *payload.Object_GetTimestampRequest, opts ...grpc.CallOption) (*payload.Object_Timestamp, error) {
+	out := new(payload.Object_Timestamp)
+	err := c.cc.Invoke(ctx, "/core.v1.Agent/GetTimestamp", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 // AgentServer is the server API for Agent service.
 // All implementations must embed UnimplementedAgentServer
 // for forward compatibility
@@ -108,6 +119,8 @@ type AgentServer interface {
 	CreateAndSaveIndex(context.Context, *payload.Control_CreateIndexRequest) (*payload.Empty, error)
 	// Represent the RPC to get the agent index information.
 	IndexInfo(context.Context, *payload.Empty) (*payload.Info_Index_Count, error)
+	// Represent the RPC to get the vector metadata. This RPC is mainly used for index correction process
+	GetTimestamp(context.Context, *payload.Object_GetTimestampRequest) (*payload.Object_Timestamp, error)
 	mustEmbedUnimplementedAgentServer()
 }
 
@@ -127,6 +140,9 @@ func (UnimplementedAgentServer) CreateAndSaveIndex(context.Context, *payload.Con
 func (UnimplementedAgentServer) IndexInfo(context.Context, *payload.Empty) (*payload.Info_Index_Count, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method IndexInfo not implemented")
 }
+func (UnimplementedAgentServer) GetTimestamp(context.Context, *payload.Object_GetTimestampRequest) (*payload.Object_Timestamp, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method GetTimestamp not implemented")
+}
 func (UnimplementedAgentServer) mustEmbedUnimplementedAgentServer() {}
 
 // UnsafeAgentServer may be embedded to opt out of forward compatibility for this service.
@@ -212,6 +228,24 @@ func _Agent_IndexInfo_Handler(srv interface{}, ctx context.Context, dec func(int
 	return interceptor(ctx, in, info, handler)
 }
 
+func _Agent_GetTimestamp_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(payload.Object_GetTimestampRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(AgentServer).GetTimestamp(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/core.v1.Agent/GetTimestamp",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(AgentServer).GetTimestamp(ctx, req.(*payload.Object_GetTimestampRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 // Agent_ServiceDesc is the grpc.ServiceDesc for Agent service.
 // It's only intended for direct use with grpc.RegisterService,
 // and not to be introspected or modified (even as a copy)
@@ -235,6 +269,10 @@ var Agent_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "IndexInfo",
 			Handler:    _Agent_IndexInfo_Handler,
 		},
+		{
+			MethodName: "GetTimestamp",
+			Handler:    _Agent_GetTimestamp_Handler,
+		},
 	},
 	Streams:  []grpc.StreamDesc{},
 	Metadata: "apis/proto/v1/agent/core/agent.proto",

apis/grpc/v1/filter/egress/egress_filter_vtproto.pb.go

@@ -66,7 +66,7 @@ func (c *filterClient) FilterDistance(ctx context.Context, in *payload.Object_Di
 }
 
 func (c *filterClient) FilterVector(ctx context.Context, in *payload.Object_Vector, opts ...grpc.CallOption) (*payload.Object_Vector, error) {
-	out := payload.Object_VectorFromVTPool()
+	out := new(payload.Object_Vector)
 	err := c.cc.Invoke(ctx, "/filter.egress.v1.Filter/FilterVector", in, out, opts...)
 	if err != nil {
 		return nil, err
@@ -127,7 +127,7 @@ func _Filter_FilterDistance_Handler(srv interface{}, ctx context.Context, dec fu
 }
 
 func _Filter_FilterVector_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
-	in := payload.Object_VectorFromVTPool()
+	in := new(payload.Object_Vector)
 	if err := dec(in); err != nil {
 		return nil, err
 	}

apis/grpc/v1/filter/ingress/ingress_filter_vtproto.pb.go

@@ -57,7 +57,7 @@ func NewFilterClient(cc grpc.ClientConnInterface) FilterClient {
 }
 
 func (c *filterClient) GenVector(ctx context.Context, in *payload.Object_Blob, opts ...grpc.CallOption) (*payload.Object_Vector, error) {
-	out := payload.Object_VectorFromVTPool()
+	out := new(payload.Object_Vector)
 	err := c.cc.Invoke(ctx, "/filter.ingress.v1.Filter/GenVector", in, out, opts...)
 	if err != nil {
 		return nil, err
@@ -66,7 +66,7 @@ func (c *filterClient) GenVector(ctx context.Context, in *payload.Object_Blob, o
 }
 
 func (c *filterClient) FilterVector(ctx context.Context, in *payload.Object_Vector, opts ...grpc.CallOption) (*payload.Object_Vector, error) {
-	out := payload.Object_VectorFromVTPool()
+	out := new(payload.Object_Vector)
 	err := c.cc.Invoke(ctx, "/filter.ingress.v1.Filter/FilterVector", in, out, opts...)
 	if err != nil {
 		return nil, err
@@ -127,7 +127,7 @@ func _Filter_GenVector_Handler(srv interface{}, ctx context.Context, dec func(in
 }
 
 func _Filter_FilterVector_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
-	in := payload.Object_VectorFromVTPool()
+	in := new(payload.Object_Vector)
 	if err := dec(in); err != nil {
 		return nil, err
 	}