[patch] [agent-NGT, sidecar] Improve S3 backup/recover behavior

.github/workflows/test.yml

@@ -25,7 +25,7 @@ jobs:
       - name: Check out code.
         uses: actions/checkout@v1
         with:
-          fetch-depth: 1
+          fetch-depth: 10
       - name: Run tests for cmd packages / tparse
         run: |
           make test/cmd/tparse | tee tparse.txt || cat tparse.txt

charts/vald/values.yaml

@@ -1080,7 +1080,7 @@ agent:
   revisionHistoryLimit: 2
   # @schema {"name": "agent.terminationGracePeriodSeconds", "type": "integer", "minimum": 0}
   # agent.terminationGracePeriodSeconds -- duration in seconds pod needs to terminate gracefully
-  terminationGracePeriodSeconds: 120
+  terminationGracePeriodSeconds: 600
   # @schema {"name": "agent.podManagementPolicy", "type": "string", "enum": ["OrderedReady", "Parallel"]}
   # agent.podManagementPolicy -- pod management policy: OrderedReady or Parallel
   podManagementPolicy: OrderedReady
@@ -1266,6 +1266,19 @@ agent:
     # @schema {"name": "agent.ngt.enable_in_memory_mode", "type": "boolean"}
     # agent.ngt.enable_in_memory_mode -- in-memory mode enabled
     enable_in_memory_mode: true
+    # @schema {"name": "agent.ngt.default_pool_size", "type": "integer"}
+    # agent.ngt.default_pool_size -- default create index batch pool size
+    default_pool_size: 10000
+    # @schema {"name": "agent.ngt.min_load_index_timeout", "type": "string"}
+    # agent.ngt.min_load_index_timeout -- minimum duration of load index timeout
+    min_load_index_timeout: 3m
+    # @schema {"name": "agent.ngt.max_load_index_timeout", "type": "string"}
+    # agent.ngt.max_load_index_timeout -- maximum duration of load index timeout
+    max_load_index_timeout: 10m
+    # @schema {"name": "agent.ngt.load_index_timeout_factor", "type": "string"}
+    # agent.ngt.load_index_timeout_factor -- a factor of load index timeout.
+    # timeout duration will be calculated by (index count to be loaded) * (factor).
+    load_index_timeout_factor: 1ms
   # @schema {"name": "agent.sidecar", "type": "object"}
   sidecar:
     # @schema {"name": "agent.sidecar.enabled", "type": "boolean"}
@@ -1372,12 +1385,18 @@ agent:
         memory: 100Mi
     # @schema {"name": "agent.sidecar.config", "type": "object"}
     config:
+      # @schema {"name": "agent.sidecar.config.watch_enabled", "type": "boolean"}
+      # agent.sidecar.config.watch_enabled -- auto backup triggered by file changes is enabled
+      watch_enabled: true
+      # @schema {"name": "agent.sidecar.config.auto_backup_enabled", "type": "boolean"}
+      # agent.sidecar.config.auto_backup_enabled -- auto backup triggered by timer is enabled
+      auto_backup_enabled: true
       # @schema {"name": "agent.sidecar.config.auto_backup_duration", "type": "string"}
       # agent.sidecar.config.auto_backup_duration -- auto backup duration
-      auto_backup_duration: 10m
+      auto_backup_duration: 24h
       # @schema {"name": "agent.sidecar.config.post_stop_timeout", "type": "string"}
-      # agent.sidecar.config.post_stop_timeout -- timeout duration for file changing during post stop
-      post_stop_timeout: 20s
+      # agent.sidecar.config.post_stop_timeout -- timeout for observing file changes during post stop
+      post_stop_timeout: 2m
       # @schema {"name": "agent.sidecar.config.filename", "type": "string"}
       # agent.sidecar.config.filename -- backup filename
       filename: _MY_POD_NAME_

dockers/agent/core/ngt/Dockerfile

@@ -19,6 +19,7 @@ FROM vdaas/vald-base:latest AS builder
 ENV ORG vdaas
 ENV REPO vald
 ENV PKG agent/core/ngt
+ENV PKG_INTERNAL agent/internal
 ENV APP_NAME ngt
 
 WORKDIR ${GOPATH}/src/github.com/${ORG}/${REPO}/internal
@@ -30,6 +31,9 @@ COPY apis/grpc .
 WORKDIR ${GOPATH}/src/github.com/${ORG}/${REPO}/pkg/${PKG}
 COPY pkg/${PKG} .
 
+WORKDIR ${GOPATH}/src/github.com/${ORG}/${REPO}/pkg/${PKG_INTERNAL}
+COPY pkg/${PKG_INTERNAL} .
+
 WORKDIR ${GOPATH}/src/github.com/${ORG}/${REPO}/cmd/${PKG}
 COPY cmd/${PKG} .
 

dockers/agent/sidecar/Dockerfile

@@ -19,6 +19,7 @@ FROM vdaas/vald-base:latest AS builder
 ENV ORG vdaas
 ENV REPO vald
 ENV PKG agent/sidecar
+ENV PKG_INTERNAL agent/internal
 ENV APP_NAME sidecar
 
 WORKDIR ${GOPATH}/src/github.com/${ORG}/${REPO}/internal
@@ -30,6 +31,9 @@ COPY apis/grpc .
 WORKDIR ${GOPATH}/src/github.com/${ORG}/${REPO}/pkg/${PKG}
 COPY pkg/${PKG} .
 
+WORKDIR ${GOPATH}/src/github.com/${ORG}/${REPO}/pkg/${PKG_INTERNAL}
+COPY pkg/${PKG_INTERNAL} .
+
 WORKDIR ${GOPATH}/src/github.com/${ORG}/${REPO}/cmd/${PKG}
 COPY cmd/${PKG} .
 

internal/compress/compress.go

@@ -22,6 +22,6 @@ import "io"
 type Compressor interface {
 	CompressVector(vector []float32) (bytes []byte, err error)
 	DecompressVector(bytes []byte) (vector []float32, err error)
-	Reader(src io.Reader) (io.Reader, error)
+	Reader(src io.ReadCloser) (io.ReadCloser, error)
 	Writer(dst io.WriteCloser) (io.WriteCloser, error)
 }

internal/compress/gob.go

@@ -60,7 +60,7 @@ func (g *gobCompressor) DecompressVector(bs []byte) ([]float32, error) {
 	return vector, nil
 }
 
-func (g *gobCompressor) Reader(src io.Reader) (io.Reader, error) {
+func (g *gobCompressor) Reader(src io.ReadCloser) (io.ReadCloser, error) {
 	return &gobReader{
 		src:     src,
 		decoder: gob.NewDecoder(src),
@@ -75,7 +75,7 @@ func (g *gobCompressor) Writer(dst io.WriteCloser) (io.WriteCloser, error) {
 }
 
 type gobReader struct {
-	src     io.Reader
+	src     io.ReadCloser
 	decoder *gob.Decoder
 }
 
@@ -88,6 +88,10 @@ func (gr *gobReader) Read(p []byte) (n int, err error) {
 	return len(p), nil
 }
 
+func (gr *gobReader) Close() error {
+	return gr.src.Close()
+}
+
 type gobWriter struct {
 	dst     io.WriteCloser
 	encoder *gob.Encoder

internal/compress/gob_test.go

@@ -121,7 +121,7 @@ func TestNewGob(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(tt *testing.T) {
-			defer goleak.VerifyNone(t)
+			defer goleak.VerifyNone(tt)
 			if test.beforeFunc != nil {
 				test.beforeFunc(test.args)
 			}
@@ -197,7 +197,7 @@ func Test_gobCompressor_CompressVector(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(tt *testing.T) {
-			defer goleak.VerifyNone(t)
+			defer goleak.VerifyNone(tt)
 			if test.beforeFunc != nil {
 				test.beforeFunc(test.args)
 			}
@@ -274,7 +274,7 @@ func Test_gobCompressor_DecompressVector(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(tt *testing.T) {
-			defer goleak.VerifyNone(t)
+			defer goleak.VerifyNone(tt)
 			if test.beforeFunc != nil {
 				test.beforeFunc(test.args)
 			}
@@ -297,22 +297,22 @@ func Test_gobCompressor_DecompressVector(t *testing.T) {
 
 func Test_gobCompressor_Reader(t *testing.T) {
 	type args struct {
-		src io.Reader
+		src io.ReadCloser
 	}
 	type want struct {
-		want io.Reader
+		want io.ReadCloser
 		err  error
 	}
 	type test struct {
 		name       string
 		args       args
 		g          *gobCompressor
 		want       want
-		checkFunc  func(want, io.Reader, error) error
+		checkFunc  func(want, io.ReadCloser, error) error
 		beforeFunc func(args)
 		afterFunc  func(args)
 	}
-	defaultCheckFunc := func(w want, got io.Reader, err error) error {
+	defaultCheckFunc := func(w want, got io.ReadCloser, err error) error {
 		if !errors.Is(err, w.err) {
 			return errors.Errorf("got error = %v, want %v", err, w.err)
 		}
@@ -351,7 +351,7 @@ func Test_gobCompressor_Reader(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(tt *testing.T) {
-			defer goleak.VerifyNone(t)
+			defer goleak.VerifyNone(tt)
 			if test.beforeFunc != nil {
 				test.beforeFunc(test.args)
 			}
@@ -428,7 +428,7 @@ func Test_gobCompressor_Writer(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(tt *testing.T) {
-			defer goleak.VerifyNone(t)
+			defer goleak.VerifyNone(tt)
 			if test.beforeFunc != nil {
 				test.beforeFunc(test.args)
 			}
@@ -454,7 +454,7 @@ func Test_gobReader_Read(t *testing.T) {
 		p []byte
 	}
 	type fields struct {
-		src     io.Reader
+		src     io.ReadCloser
 		decoder *gob.Decoder
 	}
 	type want struct {
@@ -517,7 +517,7 @@ func Test_gobReader_Read(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(tt *testing.T) {
-			defer goleak.VerifyNone(t)
+			defer goleak.VerifyNone(tt)
 			if test.beforeFunc != nil {
 				test.beforeFunc(test.args)
 			}
@@ -541,6 +541,84 @@ func Test_gobReader_Read(t *testing.T) {
 	}
 }
 
+func Test_gobReader_Close(t *testing.T) {
+	type fields struct {
+		src     io.ReadCloser
+		decoder *gob.Decoder
+	}
+	type want struct {
+		err error
+	}
+	type test struct {
+		name       string
+		fields     fields
+		want       want
+		checkFunc  func(want, error) error
+		beforeFunc func()
+		afterFunc  func()
+	}
+	defaultCheckFunc := func(w want, err error) error {
+		if !errors.Is(err, w.err) {
+			return errors.Errorf("got error = %v, want %v", err, w.err)
+		}
+		return nil
+	}
+	tests := []test{
+		// TODO test cases
+		/*
+		   {
+		       name: "test_case_1",
+		       fields: fields {
+		           src: nil,
+		           decoder: nil,
+		       },
+		       want: want{},
+		       checkFunc: defaultCheckFunc,
+		   },
+		*/
+
+		// TODO test cases
+		/*
+		   func() test {
+		       return test {
+		           name: "test_case_2",
+		           fields: fields {
+		           src: nil,
+		           decoder: nil,
+		           },
+		           want: want{},
+		           checkFunc: defaultCheckFunc,
+		       }
+		   }(),
+		*/
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(tt *testing.T) {
+			defer goleak.VerifyNone(tt)
+			if test.beforeFunc != nil {
+				test.beforeFunc()
+			}
+			if test.afterFunc != nil {
+				defer test.afterFunc()
+			}
+			if test.checkFunc == nil {
+				test.checkFunc = defaultCheckFunc
+			}
+			gr := &gobReader{
+				src:     test.fields.src,
+				decoder: test.fields.decoder,
+			}
+
+			err := gr.Close()
+			if err := test.checkFunc(test.want, err); err != nil {
+				tt.Errorf("error = %v", err)
+			}
+
+		})
+	}
+}
+
 func Test_gobWriter_Write(t *testing.T) {
 	type args struct {
 		p []byte
@@ -609,7 +687,7 @@ func Test_gobWriter_Write(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(tt *testing.T) {
-			defer goleak.VerifyNone(t)
+			defer goleak.VerifyNone(tt)
 			if test.beforeFunc != nil {
 				test.beforeFunc(test.args)
 			}
@@ -687,7 +765,7 @@ func Test_gobWriter_Close(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(tt *testing.T) {
-			defer goleak.VerifyNone(t)
+			defer goleak.VerifyNone(tt)
 			if test.beforeFunc != nil {
 				test.beforeFunc()
 			}

internal/compress/gzip.go

@@ -87,8 +87,16 @@ func (g *gzipCompressor) DecompressVector(bs []byte) ([]float32, error) {
 	return vec, nil
 }
 
-func (g *gzipCompressor) Reader(src io.Reader) (io.Reader, error) {
-	return gzip.NewReader(src)
+func (g *gzipCompressor) Reader(src io.ReadCloser) (io.ReadCloser, error) {
+	r, err := gzip.NewReader(src)
+	if err != nil {
+		return nil, err
+	}
+
+	return &gzipReader{
+		src: src,
+		r:   r,
+	}, nil
 }
 
 func (g *gzipCompressor) Writer(dst io.WriteCloser) (io.WriteCloser, error) {
@@ -103,6 +111,24 @@ func (g *gzipCompressor) Writer(dst io.WriteCloser) (io.WriteCloser, error) {
 	}, nil
 }
 
+type gzipReader struct {
+	src io.ReadCloser
+	r   io.ReadCloser
+}
+
+func (g *gzipReader) Read(p []byte) (n int, err error) {
+	return g.r.Read(p)
+}
+
+func (g *gzipReader) Close() (err error) {
+	err = g.r.Close()
+	if err != nil {
+		return errors.Wrap(g.src.Close(), err.Error())
+	}
+
+	return g.src.Close()
+}
+
 type gzipWriter struct {
 	dst io.WriteCloser
 	w   io.WriteCloser