change grpc bidi-stream error handling and change grpc API interface

apis/proto/v1/filter/egress/egress_filter.proto

@@ -29,5 +29,5 @@ service EgressFilter {
 
   rpc Filter(payload.v1.Object.Distance) returns (payload.v1.Object.Distance) {}
   rpc StreamFilter(stream payload.v1.Object.Distance)
-      returns (stream payload.v1.Object.Distance) {}
+      returns (stream payload.v1.Object.StreamDistance) {}
 }

apis/proto/v1/filter/ingress/ingress_filter.proto

@@ -35,7 +35,7 @@ service IngressFilter {
     };
   }
   rpc StreamGenVector(stream payload.v1.Object.Blob)
-      returns (stream payload.v1.Object.Vector) {}
+      returns (stream payload.v1.Object.StreamVector) {}
 
   rpc FilterVector(payload.v1.Object.Vector) returns (payload.v1.Object.Vector) {
     option (google.api.http) = {
@@ -45,5 +45,5 @@ service IngressFilter {
 
   }
   rpc StreamFilterVector(stream payload.v1.Object.Vector)
-      returns (stream payload.v1.Object.Vector) {}
+      returns (stream payload.v1.Object.StreamVector) {}
 }

apis/proto/v1/gateway/vald/vald.proto

@@ -87,7 +87,8 @@ service Vald {
     option (google.api.http).delete = "/remove/{id}";
 
   }
-  rpc StreamRemove(stream payload.v1.Object.ID) returns (stream payload.v1.Object.Location) {}
+  rpc StreamRemove(stream payload.v1.Object.ID) 
+      returns (stream payload.v1.Object.Location) {}
   rpc MultiRemove(payload.v1.Object.IDs) returns (payload.v1.Object.Locations) {}
 
   rpc GetObject(payload.v1.Object.ID) returns (payload.v1.Object.Vector) {

apis/proto/v1/manager/compressor/compressor.proto

@@ -34,10 +34,6 @@ service Backup {
     option (google.api.http).get = "/vector/{uuid}";
   }
 
-  // rpc GetVectorsByOwner(payload.v1.Backup.GetVector.Owner) returns (stream payload.v1.Backup.Vector) {
-  //   option (google.api.http).get = "/vector/{uuid}";
-  // }
-
   rpc Locations(payload.v1.Backup.Locations.Request) returns (payload.v1.Info.IPs) {
     option (google.api.http).get = "/locations/{uuid}";
   }

apis/proto/v1/payload/payload.proto

@@ -23,6 +23,7 @@ option java_multiple_files = true;
 option java_package = "org.vdaas.vald.api.v1.payload";
 option java_outer_classname = "ValdPayload";
 
+import "apis/proto/v1/errors/errors.proto";
 import "github.com/envoyproxy/protoc-gen-validate/validate/validate.proto";
 
 message Search {
@@ -64,6 +65,13 @@ message Search {
   message Responses { 
     repeated Response responses = 1; 
   }
+
+  message StreamResponse { 
+    oneof payload{
+      Response response = 1;
+      errors.Errors.RPC error = 2;
+    }
+  }
 }
 
 message Filter {
@@ -149,6 +157,13 @@ message Object {
     float distance = 2;
   }
 
+  message StreamDistance { 
+    oneof payload{
+      Distance distance = 1;
+      errors.Errors.RPC error = 2;
+    }
+  }
+
   message ID { string id = 1 [ (validate.rules).string.min_len = 1 ]; }
   message IDs { repeated string ids = 1; }
 
@@ -158,17 +173,39 @@ message Object {
   }
   message Vectors { repeated Vector vectors = 1; }
 
+  message StreamVector { 
+    oneof payload{
+      Vector vector = 1;
+      errors.Errors.RPC error = 2;
+    }
+  }
+
   message Blob {
     string id = 1 [ (validate.rules).string.min_len = 1 ];
     bytes object = 2;
   }
 
+  message StreamBlob { 
+    oneof payload{
+      Blob blob = 1;
+      errors.Errors.RPC error = 2;
+    }
+  }
+
+
   message Location {
     string name = 1;
     string uuid = 2;
     repeated string ips = 3;
   }
 
+  message StreamLocation { 
+    oneof payload{
+      Location location = 1;
+      errors.Errors.RPC error = 2;
+    }
+  }
+
   message Locations {
     repeated Location locations = 1;
   }

apis/proto/v1/vald/filter.proto

@@ -35,7 +35,7 @@ service Filter {
     };
   }
   rpc StreamSearchObject(stream payload.v1.Search.ObjectRequest)
-      returns (stream payload.v1.Search.Response) {}
+      returns (stream payload.v1.Search.StreamResponse) {}
 
   rpc InsertObject(payload.v1.Object.Blob) returns (payload.v1.Object.Location) {
     option (google.api.http) = {
@@ -45,7 +45,7 @@ service Filter {
 
   }
   rpc StreamInsertObject(stream payload.v1.Object.Blob)
-      returns (stream payload.v1.Object.Location) {}
+      returns (stream payload.v1.Object.StreamLocation) {}
   rpc MultiInsertObject(payload.v1.Object.Blob) returns (payload.v1.Object.Locations) {}
 
   rpc UpdateObject(payload.v1.Object.Blob) returns (payload.v1.Object.Location) {
@@ -55,7 +55,7 @@ service Filter {
     };
   }
   rpc StreamUpdateObject(stream payload.v1.Object.Blob)
-      returns (stream payload.v1.Object.Location) {}
+      returns (stream payload.v1.Object.StreamLocation) {}
   rpc MultiUpdateObject(payload.v1.Object.Blob) returns (payload.v1.Object.Locations) {}
 
   rpc UpsertObject(payload.v1.Object.Blob) returns (payload.v1.Object.Location) {
@@ -65,6 +65,6 @@ service Filter {
     };
   }
   rpc StreamUpsertObject(stream payload.v1.Object.Blob)
-      returns (stream payload.v1.Object.Location) {}
+      returns (stream payload.v1.Object.StreamLocation) {}
   rpc MultiUpsertObject(payload.v1.Object.Blob) returns (payload.v1.Object.Locations) {}
 }

apis/proto/v1/vald/insert.proto

@@ -37,7 +37,7 @@ service Insert {
   }
 
   rpc StreamInsert(stream payload.v1.Insert.Request)
-      returns (stream payload.v1.Object.Location) {}
+      returns (stream payload.v1.Object.StreamLocation) {}
 
   rpc MultiInsert(payload.v1.Insert.MultiRequest) returns (payload.v1.Object.Locations) {
     option (google.api.http) = {

apis/proto/v1/vald/object.proto

@@ -36,5 +36,5 @@ service Object {
     option (google.api.http).get = "/object/{id}";
   }
   rpc StreamGetObject(stream payload.v1.Object.ID)
-      returns (stream payload.v1.Object.Vector) {}
+      returns (stream payload.v1.Object.StreamVector) {}
 }

apis/proto/v1/vald/remove.proto

@@ -29,16 +29,19 @@ import "google/api/annotations.proto";
 service Remove {
 
   rpc Remove(payload.v1.Remove.Request) returns (payload.v1.Object.Location) {
-    option (google.api.http) = {
-      post : "/remove"
-      body : "*"
-    };
+      option (google.api.http) = {
+        post : "/remove"
+        body : "*"
+      };
   }
-  rpc StreamRemove(stream payload.v1.Remove.Request) returns (stream payload.v1.Object.Location) {}
+
+  rpc StreamRemove(stream payload.v1.Remove.Request) 
+      returns (stream payload.v1.Object.StreamLocation) {}
+
   rpc MultiRemove(payload.v1.Remove.MultiRequest) returns (payload.v1.Object.Locations) {
-    option (google.api.http) = {
-      post : "/remove/multiple"
-      body : "*"
-    };
+      option (google.api.http) = {
+        post : "/remove/multiple"
+        body : "*"
+      };
   }
 }

apis/proto/v1/vald/search.proto

@@ -44,10 +44,10 @@ service Search {
   }
 
   rpc StreamSearch(stream payload.v1.Search.Request)
-      returns (stream payload.v1.Search.Response) {}
+      returns (stream payload.v1.Search.StreamResponse) {}
 
   rpc StreamSearchByID(stream payload.v1.Search.IDRequest)
-      returns (stream payload.v1.Search.Response) {}
+      returns (stream payload.v1.Search.StreamResponse) {}
 
   rpc MultiSearch(payload.v1.Search.MultiRequest)
     returns (payload.v1.Search.Responses) {

apis/proto/v1/vald/update.proto

@@ -37,7 +37,7 @@ service Update {
   }
 
   rpc StreamUpdate(stream payload.v1.Update.Request)
-      returns (stream payload.v1.Object.Location) {}
+      returns (stream payload.v1.Object.StreamLocation) {}
 
   rpc MultiUpdate(payload.v1.Update.MultiRequest) returns (payload.v1.Object.Locations) {
     option (google.api.http) = {

apis/proto/v1/vald/upsert.proto

@@ -37,7 +37,7 @@ service Upsert {
   }
 
   rpc StreamUpsert(stream payload.v1.Upsert.Request)
-      returns (stream payload.v1.Object.Location) {}
+      returns (stream payload.v1.Object.StreamLocation) {}
 
   rpc MultiUpsert(payload.v1.Upsert.MultiRequest) returns (payload.v1.Object.Locations) {
     option (google.api.http) = {

go.mod

@@ -5,7 +5,7 @@ go 1.15
 replace (
 	cloud.google.com/go => cloud.google.com/go v0.75.0
 	github.com/Azure/go-autorest => github.com/Azure/go-autorest v14.2.0+incompatible
-	github.com/aws/aws-sdk-go => github.com/aws/aws-sdk-go v1.36.24
+	github.com/aws/aws-sdk-go => github.com/aws/aws-sdk-go v1.36.26
 	github.com/boltdb/bolt => github.com/boltdb/bolt v1.3.1
 	github.com/chzyer/logex => github.com/chzyer/logex v1.1.11-0.20170329064859-445be9e134b2
 	github.com/coreos/etcd => go.etcd.io/etcd v3.3.25+incompatible
@@ -16,21 +16,21 @@ replace (
 	github.com/gogo/googleapis => github.com/gogo/googleapis v1.4.0
 	github.com/gogo/protobuf => github.com/gogo/protobuf v1.3.2
 	github.com/google/go-cmp => github.com/google/go-cmp v0.5.4
-	github.com/google/pprof => github.com/google/pprof v0.0.0-20210106224343-284c741746fe
+	github.com/google/pprof => github.com/google/pprof v0.0.0-20210113181839-3dbeb6209700
 	github.com/googleapis/gnostic => github.com/googleapis/gnostic v0.5.3
 	github.com/gophercloud/gophercloud => github.com/gophercloud/gophercloud v0.15.0
 	github.com/gorilla/websocket => github.com/gorilla/websocket v1.4.2
 	github.com/hailocab/go-hostpool => github.com/monzo/go-hostpool v0.0.0-20200724120130-287edbb29340
-	github.com/klauspost/compress => github.com/klauspost/compress v1.11.7-0.20210108121608-b45090ed3abc
+	github.com/klauspost/compress => github.com/klauspost/compress v1.11.7-0.20210113214823-85d8ebff245c
 	github.com/tensorflow/tensorflow => github.com/tensorflow/tensorflow v2.1.2+incompatible
 	golang.org/x/crypto => golang.org/x/crypto v0.0.0-20201221181555-eec23a3978ad
-	google.golang.org/grpc => google.golang.org/grpc v1.34.1
+	google.golang.org/grpc => google.golang.org/grpc v1.35.0
 	google.golang.org/protobuf => google.golang.org/protobuf v1.25.0
-	k8s.io/api => k8s.io/api v0.20.1
-	k8s.io/apiextensions-apiserver => k8s.io/apiextensions-apiserver v0.20.1
-	k8s.io/apimachinery => k8s.io/apimachinery v0.20.1
-	k8s.io/client-go => k8s.io/client-go v0.20.1
-	k8s.io/metrics => k8s.io/metrics v0.20.1
+	k8s.io/api => k8s.io/api v0.20.2
+	k8s.io/apiextensions-apiserver => k8s.io/apiextensions-apiserver v0.20.2
+	k8s.io/apimachinery => k8s.io/apimachinery v0.20.2
+	k8s.io/client-go => k8s.io/client-go v0.20.2
+	k8s.io/metrics => k8s.io/metrics v0.20.2
 	sigs.k8s.io/controller-runtime => sigs.k8s.io/controller-runtime v0.7.0
 )
 
@@ -70,16 +70,16 @@ require (
 	go.uber.org/goleak v1.1.10
 	golang.org/x/net v0.0.0-20201224014010-6772e930b67b
 	golang.org/x/sync v0.0.0-20201207232520-09787c993a3a
-	golang.org/x/sys v0.0.0-20210110051926-789bb1bd4061
+	golang.org/x/sys v0.0.0-20210113181707-4bcb84eeeb78
 	gonum.org/v1/hdf5 v0.0.0-20200504100616-496fefe91614
 	gonum.org/v1/plot v0.8.1
 	google.golang.org/api v0.36.0
-	google.golang.org/genproto v0.0.0-20210111234610-22ae2b108f89
+	google.golang.org/genproto v0.0.0-20210113195801-ae06605f4595
 	google.golang.org/grpc v1.34.0
 	gopkg.in/yaml.v2 v2.4.0
-	k8s.io/api v0.20.1
-	k8s.io/apimachinery v0.20.1
-	k8s.io/client-go v0.20.1
+	k8s.io/api v0.20.2
+	k8s.io/apimachinery v0.20.2
+	k8s.io/client-go v0.20.2
 	k8s.io/metrics v0.0.0-00010101000000-000000000000
 	sigs.k8s.io/controller-runtime v0.0.0-00010101000000-000000000000
 )