Logging: Adding unique writer identity to Sink. (#4595)

logging/google/cloud/logging/_gax.py

@@ -192,7 +192,8 @@ def list_sinks(self, project, page_size=0, page_token=None):
         return page_iterator._GAXIterator(
             self._client, page_iter, _item_to_sink)
 
-    def sink_create(self, project, sink_name, filter_, destination):
+    def sink_create(self, project, sink_name, filter_, destination,
+                    unique_writer_identity=False):
         """API call:  create a sink resource.
 
         See
@@ -211,13 +212,23 @@ def sink_create(self, project, sink_name, filter_, destination):
         :type destination: str
         :param destination: destination URI for the entries exported by
                             the sink.
+
+        :type unique_writer_identity: bool
+        :param unique_writer_identity: (Optional) determines the kind of
+                                       IAM identity returned as
+                                       writer_identity in the new sink.
         """
         options = None
         parent = 'projects/%s' % (project,)
         sink_pb = LogSink(name=sink_name, filter=filter_,
                           destination=destination)
         try:
-            self._gax_api.create_sink(parent, sink_pb, options=options)
+            self._gax_api.create_sink(
+                parent,
+                sink_pb,
+                unique_writer_identity=unique_writer_identity,
+                options=options,
+            )
         except GaxError as exc:
             if exc_to_code(exc.cause) == StatusCode.FAILED_PRECONDITION:
                 path = 'projects/%s/sinks/%s' % (project, sink_name)

logging/google/cloud/logging/_http.py

@@ -229,7 +229,8 @@ def list_sinks(self, project, page_size=None, page_token=None):
             page_token=page_token,
             extra_params=extra_params)
 
-    def sink_create(self, project, sink_name, filter_, destination):
+    def sink_create(self, project, sink_name, filter_, destination,
+                    unique_writer_identity=False):
         """API call:  create a sink resource.
 
         See
@@ -248,14 +249,25 @@ def sink_create(self, project, sink_name, filter_, destination):
         :type destination: str
         :param destination: destination URI for the entries exported by
                             the sink.
+
+        :type unique_writer_identity: bool
+        :param unique_writer_identity: (Optional) determines the kind of
+                                       IAM identity returned as
+                                       writer_identity in the new sink.
         """
         target = '/projects/%s/sinks' % (project,)
         data = {
             'name': sink_name,
             'filter': filter_,
             'destination': destination,
         }
-        self.api_request(method='POST', path=target, data=data)
+        query_params = {'uniqueWriterIdentity': unique_writer_identity}
+        self.api_request(
+            method='POST',
+            path=target,
+            data=data,
+            query_params=query_params,
+        )
 
     def sink_get(self, project, sink_name):
         """API call:  retrieve a sink resource.

logging/google/cloud/logging/sink.py

@@ -39,12 +39,19 @@ class Sink(object):
     :type client: :class:`google.cloud.logging.client.Client`
     :param client: A client which holds credentials and project configuration
                    for the sink (which requires a project).
+
+    :type unique_writer_identity: bool
+    :param unique_writer_identity: (Optional) determines the kind of
+                                   IAM identity returned as
+                                   writer_identity in the new sink.
     """
-    def __init__(self, name, filter_=None, destination=None, client=None):
+    def __init__(self, name, filter_=None, destination=None, client=None,
+                 unique_writer_identity=False):
         self.name = name
         self.filter_ = filter_
         self.destination = destination
         self._client = client
+        self._unique_writer_identity = unique_writer_identity
 
     @property
     def client(self):
@@ -116,7 +123,9 @@ def create(self, client=None):
         """
         client = self._require_client(client)
         client.sinks_api.sink_create(
-            self.project, self.name, self.filter_, self.destination)
+            self.project, self.name, self.filter_, self.destination,
+            unique_writer_identity=self._unique_writer_identity,
+        )
 
     def exists(self, client=None):
         """API call:  test for the existence of the sink via a GET request

logging/tests/unit/test__gax.py

@@ -737,14 +737,34 @@ def test_sink_create_ok(self):
         api.sink_create(
             self.PROJECT, self.SINK_NAME, self.FILTER, self.DESTINATION_URI)
 
-        parent, sink, options = (
+        parent, sink, options, unique_writer_identity = (
             gax_api._create_sink_called_with)
         self.assertEqual(parent, self.PROJECT_PATH)
         self.assertIsInstance(sink, LogSink)
         self.assertEqual(sink.name, self.SINK_NAME)
         self.assertEqual(sink.filter, self.FILTER)
         self.assertEqual(sink.destination, self.DESTINATION_URI)
         self.assertIsNone(options)
+        self.assertFalse(unique_writer_identity)
+
+    def test_sink_create_with_unique_writer_identity(self):
+        from google.cloud.proto.logging.v2.logging_config_pb2 import LogSink
+
+        gax_api = _GAXSinksAPI()
+        api = self._make_one(gax_api, None)
+        api.sink_create(
+            self.PROJECT, self.SINK_NAME, self.FILTER, self.DESTINATION_URI,
+            unique_writer_identity=True,
+        )
+        parent, sink, options, unique_writer_identity = (
+            gax_api._create_sink_called_with)
+        self.assertEqual(parent, self.PROJECT_PATH)
+        self.assertIsInstance(sink, LogSink)
+        self.assertEqual(sink.name, self.SINK_NAME)
+        self.assertEqual(sink.filter, self.FILTER)
+        self.assertEqual(sink.destination, self.DESTINATION_URI)
+        self.assertIsNone(options)
+        self.assertTrue(unique_writer_identity)
 
     def test_sink_get_error(self):
         from google.cloud.exceptions import NotFound
@@ -1462,10 +1482,10 @@ def list_sinks(self, parent, page_size, options):
         self._list_sinks_called_with = parent, page_size, options
         return self._list_sinks_response
 
-    def create_sink(self, parent, sink, options):
+    def create_sink(self, parent, sink, options, unique_writer_identity=False):
         from google.gax.errors import GaxError
 
-        self._create_sink_called_with = parent, sink, options
+        self._create_sink_called_with = parent, sink, options, unique_writer_identity
         if self._random_gax_error:
             raise GaxError('error')
         if self._create_sink_conflict:

logging/tests/unit/test__http.py

@@ -476,6 +476,33 @@ def test_sink_create_ok(self):
         self.assertEqual(conn._called_with['path'], path)
         self.assertEqual(conn._called_with['data'], SENT)
 
+    def test_sink_create_unique_writer_identity(self):
+        sent = {
+            'name': self.SINK_NAME,
+            'filter': self.FILTER,
+            'destination': self.DESTINATION_URI,
+        }
+
+        conn = _Connection({})
+        client = _Client(conn)
+        api = self._make_one(client)
+
+        api.sink_create(
+            self.PROJECT,
+            self.SINK_NAME,
+            self.FILTER,
+            self.DESTINATION_URI,
+            unique_writer_identity=True,
+        )
+        path = '/projects/%s/sinks' % (self.PROJECT,)
+        expected = {
+            'method': 'POST',
+            'path': path,
+            'data': sent,
+            'query_params': {'uniqueWriterIdentity': True},
+        }
+        self.assertEqual(conn._called_with, expected)
+
     def test_sink_get_miss(self):
         from google.cloud.exceptions import NotFound
 

logging/tests/unit/test_sink.py

@@ -102,7 +102,14 @@ def test_create_w_bound_client(self):
 
         self.assertEqual(
             api._sink_create_called_with,
-            (self.PROJECT, self.SINK_NAME, self.FILTER, self.DESTINATION_URI))
+            (
+                self.PROJECT,
+                self.SINK_NAME,
+                self.FILTER,
+                self.DESTINATION_URI,
+                False,
+            ),
+        )
 
     def test_create_w_alternate_client(self):
         client1 = _Client(project=self.PROJECT)
@@ -116,7 +123,14 @@ def test_create_w_alternate_client(self):
 
         self.assertEqual(
             api._sink_create_called_with,
-            (self.PROJECT, self.SINK_NAME, self.FILTER, self.DESTINATION_URI))
+            (
+                self.PROJECT,
+                self.SINK_NAME,
+                self.FILTER,
+                self.DESTINATION_URI,
+                False,
+            ),
+        )
 
     def test_exists_miss_w_bound_client(self):
         client = _Client(project=self.PROJECT)
@@ -255,9 +269,10 @@ def __init__(self, project):
 
 class _DummySinksAPI(object):
 
-    def sink_create(self, project, sink_name, filter_, destination):
+    def sink_create(self, project, sink_name, filter_, destination,
+                    unique_writer_identity=False):
         self._sink_create_called_with = (
-            project, sink_name, filter_, destination)
+            project, sink_name, filter_, destination, unique_writer_identity)
 
     def sink_get(self, project, sink_name):
         from google.cloud.exceptions import NotFound