Make retry logic idempotent in GcsIO.delete and GcsIO.rename

----Release Notes----
[]
-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=124180161

Author: charlesccychen

google/cloud/dataflow/io/gcsio.py

@@ -139,38 +139,54 @@ def glob(self, pattern):
 
   @retry.with_exponential_backoff()  # Using retry defaults from utils/retry.py
   def delete(self, path):
-    """Deletes the given gcs object.
+    """Deletes the object at the given GCS path.
 
     Args:
       path: GCS file path pattern in the form gs://<bucket>/<name>.
     """
     bucket, object_path = parse_gcs_path(path)
     request = storage.StorageObjectsDeleteRequest(bucket=bucket,
                                                   object=object_path)
-    self.client.objects.Delete(request)
+    try:
+      self.client.objects.Delete(request)
+    except HttpError as http_error:
+      if http_error.status_code == 404:
+        # Return success when the file doesn't exist anymore for idempotency.
+        return
+      raise
 
   @retry.with_exponential_backoff()  # Using retry defaults from utils/retry.py
-  def rename(self, src, dst):
-    """Renames the given gcs object from src to dst.
+  def copy(self, src, dest):
+    """Copies the given GCS object from src to dest.
 
     Args:
       src: GCS file path pattern in the form gs://<bucket>/<name>.
-      dst: GCS file path pattern in the form gs://<bucket>/<name>.
+      dest: GCS file path pattern in the form gs://<bucket>/<name>.
     """
     src_bucket, src_path = parse_gcs_path(src)
-    dst_bucket, dst_path = parse_gcs_path(dst)
+    dest_bucket, dest_path = parse_gcs_path(dest)
     request = storage.StorageObjectsCopyRequest(sourceBucket=src_bucket,
                                                 sourceObject=src_path,
-                                                destinationBucket=dst_bucket,
-                                                destinationObject=dst_path)
+                                                destinationBucket=dest_bucket,
+                                                destinationObject=dest_path)
     self.client.objects.Copy(request)
-    request = storage.StorageObjectsDeleteRequest(bucket=src_bucket,
-                                                  object=src_path)
-    self.client.objects.Delete(request)
+
+  # We intentionally do not decorate this method with a retry, since the
+  # underlying copy and delete operations are already idempotent operations
+  # protected by retry decorators.
+  def rename(self, src, dest):
+    """Renames the given GCS object from src to dest.
+
+    Args:
+      src: GCS file path pattern in the form gs://<bucket>/<name>.
+      dest: GCS file path pattern in the form gs://<bucket>/<name>.
+    """
+    self.copy(src, dest)
+    self.delete(src)
 
   @retry.with_exponential_backoff()  # Using retry defaults from utils/retry.py
   def exists(self, path):
-    """Returns whether the given gcs object exists.
+    """Returns whether the given GCS object exists.
 
     Args:
       path: GCS file path pattern in the form gs://<bucket>/<name>.

google/cloud/dataflow/io/gcsio_test.py

@@ -21,8 +21,11 @@
 import threading
 import unittest
 
-from google.cloud.dataflow.io import gcsio
 
+import httplib2
+
+from google.cloud.dataflow.io import gcsio
+from apitools.base.py.exceptions import HttpError
 from google.cloud.dataflow.internal.clients import storage
 
 
@@ -53,14 +56,24 @@ class FakeGcsObjects(object):
 
   def __init__(self):
     self.files = {}
+    # Store the last generation used for a given object name.  Note that this
+    # has to persist even past the deletion of the object.
+    self.last_generation = {}
     self.list_page_tokens = {}
 
   def add_file(self, f):
     self.files[(f.bucket, f.object)] = f
+    self.last_generation[(f.bucket, f.object)] = f.generation
 
   def get_file(self, bucket, obj):
     return self.files.get((bucket, obj), None)
 
+  def delete_file(self, bucket, obj):
+    del self.files[(bucket, obj)]
+
+  def get_last_generation(self, bucket, obj):
+    return self.last_generation.get((bucket, obj), 0)
+
   def Get(self, get_request, download=None):  # pylint: disable=invalid-name
     f = self.get_file(get_request.bucket, get_request.object)
     if f is None:
@@ -77,10 +90,8 @@ def get_range_callback(start, end):
 
   def Insert(self, insert_request, upload=None):  # pylint: disable=invalid-name
     assert upload is not None
-    generation = 1
-    f = self.get_file(insert_request.bucket, insert_request.name)
-    if f is not None:
-      generation = f.generation + 1
+    generation = self.get_last_generation(insert_request.bucket,
+                                          insert_request.name) + 1
     f = FakeFile(insert_request.bucket, insert_request.name, '', generation)
 
     # Stream data into file.
@@ -95,6 +106,26 @@ def Insert(self, insert_request, upload=None):  # pylint: disable=invalid-name
 
     self.add_file(f)
 
+  def Copy(self, copy_request):  # pylint: disable=invalid-name
+    src_file = self.get_file(copy_request.sourceBucket,
+                             copy_request.sourceObject)
+    assert src_file is not None
+    generation = self.get_last_generation(copy_request.destinationBucket,
+                                          copy_request.destinationObject) + 1
+    dest_file = FakeFile(copy_request.destinationBucket,
+                         copy_request.destinationObject,
+                         src_file.contents, generation)
+    self.add_file(dest_file)
+
+  def Delete(self, delete_request):  # pylint: disable=invalid-name
+    # Here, we emulate the behavior of the GCS service in raising a 404 error
+    # if this object already exists.
+    if self.get_file(delete_request.bucket, delete_request.object):
+      self.delete_file(delete_request.bucket, delete_request.object)
+    else:
+      raise HttpError(httplib2.Response({'status': '404'}), '404 Not Found',
+                      'https://fake/url')
+
   def List(self, list_request):  # pylint: disable=invalid-name
     bucket = list_request.bucket
     prefix = list_request.prefix or ''
@@ -154,6 +185,58 @@ def setUp(self):
     self.client = FakeGcsClient()
     self.gcs = gcsio.GcsIO(self.client)
 
+  def test_delete(self):
+    file_name = 'gs://gcsio-test/delete_me'
+    file_size = 1024
+
+    # Test deletion of non-existent file.
+    self.gcs.delete(file_name)
+
+    self._insert_random_file(self.client, file_name, file_size)
+    self.assertTrue(gcsio.parse_gcs_path(file_name) in
+                    self.client.objects.files)
+
+    self.gcs.delete(file_name)
+
+    self.assertFalse(gcsio.parse_gcs_path(file_name) in
+                     self.client.objects.files)
+
+  def test_copy(self):
+    src_file_name = 'gs://gcsio-test/source'
+    dest_file_name = 'gs://gcsio-test/dest'
+    file_size = 1024
+    self._insert_random_file(self.client, src_file_name,
+                             file_size)
+    self.assertTrue(gcsio.parse_gcs_path(src_file_name) in
+                    self.client.objects.files)
+    self.assertFalse(gcsio.parse_gcs_path(dest_file_name) in
+                     self.client.objects.files)
+
+    self.gcs.copy(src_file_name, dest_file_name)
+
+    self.assertTrue(gcsio.parse_gcs_path(src_file_name) in
+                    self.client.objects.files)
+    self.assertTrue(gcsio.parse_gcs_path(dest_file_name) in
+                    self.client.objects.files)
+
+  def test_rename(self):
+    src_file_name = 'gs://gcsio-test/source'
+    dest_file_name = 'gs://gcsio-test/dest'
+    file_size = 1024
+    self._insert_random_file(self.client, src_file_name,
+                             file_size)
+    self.assertTrue(gcsio.parse_gcs_path(src_file_name) in
+                    self.client.objects.files)
+    self.assertFalse(gcsio.parse_gcs_path(dest_file_name) in
+                     self.client.objects.files)
+
+    self.gcs.rename(src_file_name, dest_file_name)
+
+    self.assertFalse(gcsio.parse_gcs_path(src_file_name) in
+                     self.client.objects.files)
+    self.assertTrue(gcsio.parse_gcs_path(dest_file_name) in
+                    self.client.objects.files)
+
   def test_full_file_read(self):
     file_name = 'gs://gcsio-test/full_file'
     file_size = 5 * 1024 * 1024 + 100