Feat/replication

aruna/api/dataproxy/services/v2/dataproxy_service.proto

@@ -14,22 +14,22 @@ import "google/api/visibility.proto";
 // Status: ALPHA
 //
 // Service for data replication between data-proxies
-service DataproxyService {
+service DataproxyReplicationService {
     option (google.api.api_visibility).restriction = "PROXY";
 
     // RequestReplication
     //
     // Status: ALPHA
     //
-    // Creates a replication request
-    rpc RequestReplication(RequestReplicationRequest) returns (RequestReplicationResponse) {}
+    // Creates a replication stream
+    rpc PullReplication(stream PullReplicationRequest) returns (stream PullReplicationResponse) {}
 
     // InitReplication
     //
     // Status: ALPHA
     //
     // Provides the necessary url to init replication
-    rpc InitReplication(InitReplicationRequest) returns (InitReplicationResponse) {}
+    rpc PushReplication(PushReplicationRequest) returns (PushReplicationResponse) {}
 }
 
 
@@ -96,19 +96,76 @@ service DataproxyUserService {
 }
 
 
+// ----- PullReplication -----
+// PROXY A (data) <--> PROXY B (wants data)
+// PROXY B pulls data from PROXY A
 
+// Messages (requests) from PROXY B
+message InitMessage {
+    string dataproxy_id = 1;
+    repeated string object_ids = 2;
+}
 
+message InfoAckMessage {
+    string object_id = 1;
+}
 
-message DataProxyInfo {
-    string dataproxy_id = 1;
-    int64 available_space = 2;
+message ChunkAckMessage {
+    string object_id = 1;
+    int64 chunk_idx = 2;
 }
 
-message RequestReplicationRequest {
-    DataProxyInfo info = 1;
-    bool user_initialized = 2;
+message RetryChunkMessage {
+    string object_id = 1;
+    int64 chunk_idx = 2;
 }
 
+message Empty {}
+
+message ErrorMessage {
+    oneof error {
+        RetryChunkMessage retry_chunk = 1;
+        Empty abort = 2;
+        string retry_object_id = 3;
+    }
+}
+
+message PullReplicationRequest {
+    oneof message {
+        InitMessage init_message = 1;
+        InfoAckMessage info_ack_message = 2;
+        ChunkAckMessage chunk_ack_message = 3;
+        ErrorMessage error_message = 4;
+        Empty finish_message = 5;
+    }
+}
+
+// Messages (responses) from PROXY A
+message ObjectInfo {
+    string object_id = 1;
+    int64 chunks = 2;
+    int64 raw_size  = 3;
+    repeated uint32 block_list = 4;
+    optional string extra = 5; // JSON encoded proxy specific extra fields
+}
+
+message Chunk {
+    string object_id = 1;
+    int64 chunk_idx = 2;
+    bytes data = 3;
+    string checksum = 4;
+}
+
+message PullReplicationResponse {
+    oneof message {
+        ObjectInfo object_info = 1;
+        Chunk chunk = 2; // If no ack is received, the chunk will be resent
+        Empty finish_message = 3;
+    }
+}
+
+// ---------------
+
 message DataInfo {
     string object_id = 1;
     string download_url = 2;
@@ -120,18 +177,11 @@ message DataInfos {
     repeated DataInfo data_info = 1;
 }
 
-message RequestReplicationResponse {
-    oneof response {
-        DataInfos data_infos = 1;
-        bool ack = 2; 
-    }
-}
-
-message InitReplicationRequest {
+message PushReplicationRequest {
     DataInfos data_infos = 1;
 }
 
-message InitReplicationResponse {
+message PushReplicationResponse {
     bool ack = 1;
 }
 
@@ -152,7 +202,7 @@ message PushReplicaRequest {
         string resource_id = 1;
         S3Path s3_path = 2;
     }
-    string target_location = 3;
+    string target_endpoint_id = 3;
 }
 
 message PushReplicaResponse {
@@ -167,11 +217,11 @@ message PullReplicaRequest {
 }
 
 message PullReplicaResponse {
-    string replication_id = 1;
+    string replication_id = 1; // why ?
 }
 
 message ReplicationStatusRequest {
-    string replication_id = 1;
+    string replication_id = 1; // why ?
 }
 
 enum ReplicationStatus {

aruna/api/hooks/services/v2/hooks_service.proto

@@ -85,6 +85,7 @@ service HooksService {
     rpc HookCallback(HookCallbackRequest) returns (HookCallbackResponse) {
         option (google.api.http) = {
             patch : "/v2/hooks/callback"
+	    body: "*"
         };
     }
 }

aruna/api/storage/models/v2/models.proto

@@ -121,6 +121,13 @@ enum ResourceVariant {
   RESOURCE_VARIANT_OBJECT = 4;
 }
 
+enum ReplicationStatus {
+  REPLICATION_STATUS_UNSPECIFIED = 0;
+  REPLICATION_STATUS_WAITING = 1;
+  REPLICATION_STATUS_RUNNING = 2;
+  REPLICATION_STATUS_FINISHED = 3;
+  REPLICATION_STATUS_ERROR = 4;
+}
 // ------------- USERS & PERMISSIONS -----------------------
 message User {
   reserved 2;
@@ -251,7 +258,23 @@ message DataEndpoint {
   string id = 1;
   // Hint if the objects' project 
   // is fully synced to the endpoint
-  bool full_synced = 2;
+  oneof variant {
+      FullSync full_sync = 2;
+      PartialSync partial_sync = 3;
+  }
+  optional ReplicationStatus status = 4;
+}
+
+message FullSync {
+   string project_id = 1; 
+}
+message PartialSync {
+   oneof origin {
+    string project_id = 1;
+    string collection_id = 2;
+    string dataset_id = 3;
+    string object_id = 4;
+  } 
 }
 
 message Copy {

aruna/api/storage/services/v2/data_replication_service.proto

@@ -7,9 +7,9 @@ option java_package = "com.github.ArunaStorage.java_api.aruna.api.storage.servic
 option java_outer_classname = "DataReplicationService";
 
 import "google/api/annotations.proto";
+import "aruna/api/storage/models/v2/models.proto";
 
 // DataReplicationService
-//
 // Endpoint specific methods for syncing data
 service DataReplicationService {
 
@@ -34,7 +34,7 @@ service DataReplicationService {
     rpc PartialReplicateData(PartialReplicateDataRequest)
         returns (PartialReplicateDataResponse) {
       option (google.api.http) = {
-        post : "/v2/endpoints/{endpoint_id}/replication/{resource_id}/partial"
+        post : "/v2/endpoints/{endpoint_id}/replication/partial"
         body: "*"
     };
     }
@@ -47,7 +47,7 @@ service DataReplicationService {
     rpc UpdateReplicationStatus(UpdateReplicationStatusRequest)
         returns (UpdateReplicationStatusResponse) {
       option (google.api.http) = {
-        patch : "/v2/endpoints/{endpoint_id}/replication/{resource_id}/status"
+        patch : "/v2/endpoints/{endpoint_id}/replication/{object_id}/status"
         body: "*"
       };
     }
@@ -77,36 +77,35 @@ service DataReplicationService {
     }
 }
 
-enum ReplicationStatus {
-  REPLICATION_STATUS_UNSPECIFIED = 0;
-  REPLICATION_STATUS_WAITING = 1;
-  REPLICATION_STATUS_RUNNING = 2;
-  REPLICATION_STATUS_FINISHED = 3;
-  REPLICATION_STATUS_ERROR = 4;
-}
+
 
 message ReplicateProjectDataRequest {
   string project_id = 1;
   string endpoint_id = 2;
 }
 
 message ReplicateProjectDataResponse {
-  ReplicationStatus status = 1;
+  storage.models.v2.ReplicationStatus status = 1;
 }
 
 message PartialReplicateDataRequest {
-  string resource_id = 1;
-  string endpoint_id = 2;
+    oneof data_variant {
+	string collection_id = 1;
+	string dataset_id = 2;
+	string object_id = 3;
+    }
+    string endpoint_id = 4;
 }
 
+
 message PartialReplicateDataResponse {
-  ReplicationStatus status = 1;
+  storage.models.v2.ReplicationStatus status = 1;
 }
 
 message UpdateReplicationStatusRequest {
-  string resource_id = 1;
+  string object_id = 1;
   string endpoint_id = 2;
-  ReplicationStatus status = 3;
+  storage.models.v2.ReplicationStatus status = 3;
 }
 
 message UpdateReplicationStatusResponse {}
@@ -117,7 +116,17 @@ message GetReplicationStatusRequest {
 }
 
 message GetReplicationStatusResponse {
-  ReplicationStatus status = 1;
+  repeated ReplicationInfo infos = 1;
+}
+
+message ReplicationInfo {
+    oneof resource {
+	string project_id = 1;
+	string collection_id = 2;
+	string dataset_id = 3;
+	string object_id = 4;
+    }
+    storage.models.v2.DataEndpoint endpoint_info = 5;
 }
 
 message DeleteReplicationRequest {