instrumenting job.py and adding documentation

tswast · Aug 24, 2020 · Aug 1, 2020 · Aug 3, 2020 · Aug 4, 2020 · Aug 4, 2020
commit 63e9581f90ef33f946cb2938643379ad5770d0ff
diff --git a/README.rst b/README.rst
@@ -102,3 +102,51 @@ Perform a query
 
     for row in rows:
         print(row.name)
+
+Instrumenting With OpenTelemetry
+--------------------------------
+
+This application uses OpenTelmetry to output tracing data from
+api calls to BigQuery. To enable OpenTelemetry tracing in
+the BigQuery client the following PyPi packages need to be installed:
+
+.. code-block:: console
+
+    pip install opentelemetry-api
+
+.. code-block:: console
+
+    pip install opentelemetry-sdk
+
+.. code-block:: console
+
+    pip install opentelemetry-instrumentaion
+
+.. code-block:: console
+
+    pip install opentelemetry-exporter-cloud_trace
+
+After installation, OpenTelemetry can now be used in the BigQuery
+client and in BigQuery jobs. First however an exporter must be
+specified for where the trace data will be outputted to. An example of this
+can be found here:
+
+.. code-block:: python
+
+    from opentelemetry import trace
+    from opentelemetry.sdk.trace import TracerProvider
+    from opentelemetry.sdk.trace.export import SimpleExportSpanProcessor
+    from opentelemetry.exporter.cloud_trace import CloudTraceSpanExporter
+    trace.set_tracer_provider(TracerProvider())
+    trace.get_tracer_provider().add_span_processor(
+        SimpleExportSpanProcessor(CloudTraceSpanExporter())
+    )
+
+In this example all tracing data will be published to the Google
+Cloud trace console. For more information on OpenTelemetry, please consult the `OpenTelemetry documentation`_.
+
+.. _OpenTelemetry documentation: https://opentelemetry-python.readthedocs.io
+
+
+
+
diff --git a/google/cloud/bigquery/job.py b/google/cloud/bigquery/job.py
@@ -50,6 +50,8 @@
 from google.cloud.bigquery.table import TableReference
 from google.cloud.bigquery.table import Table
 from google.cloud.bigquery.table import TimePartitioning
+from opentelemetry_tracing import SpanCreator
+
 
 _DONE_STATE = "DONE"
 _STOPPED_REASON = "stopped"
@@ -634,9 +636,12 @@ def _begin(self, client=None, retry=DEFAULT_RETRY, timeout=None):
 
         # jobs.insert is idempotent because we ensure that every new
         # job has an ID.
-        api_response = client._call_api(
-            retry, method="POST", path=path, data=self.to_api_repr(), timeout=timeout
-        )
+        attributes = {'path': path}
+        span_creator = SpanCreator()
+        with span_creator.create('BigQuery.job.begin', attributes):
+            api_response = client._call_api(
+                retry, method="POST", path=path, data=self.to_api_repr(), timeout=timeout
+            )
         self._set_properties(api_response)
 
     def exists(self, client=None, retry=DEFAULT_RETRY, timeout=None):
@@ -665,13 +670,17 @@ def exists(self, client=None, retry=DEFAULT_RETRY, timeout=None):
             extra_params["location"] = self.location
 
         try:
-            client._call_api(
-                retry,
-                method="GET",
-                path=self.path,
-                query_params=extra_params,
-                timeout=timeout,
-            )
+            span_creator = SpanCreator()
+            attributes = extra_params
+            attributes['path'] = self.path
+            with span_creator.create('BigQuery.job.exists', attributes):
+                client._call_api(
+                    retry,
+                    method="GET",
+                    path=self.path,
+                    query_params=extra_params,
+                    timeout=timeout,
+                )
         except NotFound:
             return False
         else:
@@ -698,14 +707,17 @@ def reload(self, client=None, retry=DEFAULT_RETRY, timeout=None):
         extra_params = {}
         if self.location:
             extra_params["location"] = self.location
-
-        api_response = client._call_api(
-            retry,
-            method="GET",
-            path=self.path,
-            query_params=extra_params,
-            timeout=timeout,
-        )
+        span_creator = SpanCreator()
+        attributes = extra_params
+        attributes['path'] = self.path
+        with span_creator.create('BigQuery.job.reload', attributes):
+            api_response = client._call_api(
+                retry,
+                method="GET",
+                path=self.path,
+                query_params=extra_params,
+                timeout=timeout,
+            )
         self._set_properties(api_response)
 
     def cancel(self, client=None, retry=DEFAULT_RETRY, timeout=None):
@@ -732,13 +744,18 @@ def cancel(self, client=None, retry=DEFAULT_RETRY, timeout=None):
         if self.location:
             extra_params["location"] = self.location
 
-        api_response = client._call_api(
-            retry,
-            method="POST",
-            path="{}/cancel".format(self.path),
-            query_params=extra_params,
-            timeout=timeout,
-        )
+        path = "{}/cancel".format(self.path)
+        span_creator = SpanCreator()
+        attributes = extra_params
+        attributes['path'] = path
+        with span_creator.create('BigQuery.job.cancel', attributes):
+            api_response = client._call_api(
+                retry,
+                method="POST",
+                path=path,
+                query_params=extra_params,
+                timeout=timeout,
+            )
         self._set_properties(api_response["job"])
         # The Future interface requires that we return True if the *attempt*
         # to cancel was successful.