Add the ability to view operation details from Google Pipelines API [BA-5762] (#661)

Author: rsasch

Diff for: CHANGELOG.md

@@ -1,5 +1,9 @@
 # Job Manager Change Log
 
+## v1.3.0 Release Notes
+
+### Added icon that, when clicked, will provide Google Pipelines operation details. (So far) only implemented for a Cromwell backend, and it requires `outsideAuth` to be set to `true` in the authentication section of the `capabilities_config.json` file.
+
 ## v1.2.3 Release Notes
 
 ### Added logic to avoid 500 error on Job Details page when a workflow failure section does not have a causedBy entry.
@@ -14,7 +18,7 @@
 
 ## v1.2.0 Release Notes
 
-### Added the ability – with the addition of a specific scope to the capabilities config file – to see the contents of Google Storage log files within the UI.
+### Added the ability -- with the addition of a specific scope to the capabilities config file -- to see the contents of Google Storage log files within the UI.
 
 ## v1.1.1 Release Notes
 

Diff for: api/jobs.yaml

@@ -15,6 +15,7 @@ enum: &TIMEFRAME
   - DAYS_7
   - DAYS_30
   - ALL_TIME
+
 paths:
   '/capabilities':
     get:
@@ -31,6 +32,7 @@ paths:
           $ref: '#/responses/ServerError'
       tags:
         - Capabilities
+
   '/jobs/{id}/abort':
     post:
       operationId: AbortJob
@@ -56,6 +58,7 @@ paths:
           $ref: '#/responses/ServerError'
       tags:
         - Jobs
+
   '/jobs/{id}/updateLabels':
     post:
       operationId: UpdateJobLabels
@@ -86,6 +89,7 @@ paths:
           description: Server does not implement this method.
       tags:
         - Jobs
+
   '/jobs/query':
     post:
       # TODO(calbach): Parameterize sort ordering.
@@ -112,6 +116,7 @@ paths:
           $ref: '#/responses/ServerError'
       tags:
         - Jobs
+
   '/jobs/{id}':
     get:
       operationId: GetJob
@@ -137,6 +142,7 @@ paths:
           $ref: '#/responses/ServerError'
       tags:
         - Jobs
+
   '/jobs/{id}/{task}/attempts':
     get:
       operationId: GetTaskAttempts
@@ -167,6 +173,7 @@ paths:
           $ref: '#/responses/ServerError'
       tags:
         - Jobs
+
   '/jobs/{id}/{task}/{index}/attempts':
     get:
       operationId: GetShardAttempts
@@ -202,6 +209,33 @@ paths:
           $ref: '#/responses/ServerError'
       tags:
         - Jobs
+
+  '/jobs/operationDetails':
+    get:
+      operationId: GetOperationDetails
+      summary: Call for operation details from the Google Pipelines API
+      parameters:
+        - name: job
+          description: job ID
+          required: true
+          in: query
+          type: string
+        - name: operation
+          description: operation ID
+          type: string
+          in: query
+      responses:
+        '200':
+          description: Success
+          schema:
+            $ref: '#/definitions/JobOperationResponse'
+        '404':
+          $ref: '#/responses/NotFound'
+        '500':
+          $ref: '#/responses/ServerError'
+      tags:
+        - Jobs
+
   '/aggregations':
     get:
       operationId: GetJobAggregations
@@ -232,6 +266,7 @@ paths:
           $ref: '#/responses/ServerError'
       tags:
         - Aggregation
+
   '/health':
     get:
       operationId: Health
@@ -267,6 +302,10 @@ definitions:
       forcedLogoutTime:
         type: integer
         description: Number of milliseconds for the interval of inactivity.
+      outsideAuth:
+        type: boolean
+        description: Whether or not to use an outside service for some auth-related functionality.
+        default: false
 
   DisplayField:
     description: Description of a display field
@@ -379,6 +418,15 @@ definitions:
       extensions:
         $ref: '#/definitions/ExtendedFields'
 
+  JobOperationResponse:
+    description: Google Pipelines API provided details about an operation
+    properties:
+      details:
+        type: string
+      id:
+        type: string
+        description: operation ID
+
   UpdateJobLabelsRequest:
     description: Update labels on a job
     properties:
@@ -683,6 +731,9 @@ definitions:
       callRoot:
         type: string
         description: The root directory of the task execution
+      operationId:
+        type: string
+        description: Operation Id for Google Pipelines API
       attempts:
         type: integer
         description: Number of times the task was run
@@ -726,6 +777,9 @@ definitions:
       callRoot:
         type: string
         description: The root directory of the task execution
+      operationId:
+        type: string
+        description: Operation Id for Google Pipelines API
       inputs:
         type: object
         description: Mapping of input fully qualified names to stringified values
@@ -778,6 +832,9 @@ definitions:
       callRoot:
         type: string
         description: If from a task, the root directory of the task or shard execution
+      operationId:
+        type: string
+        description: Operation Id for Google Pipelines API
       attempts:
         type: integer
         description: Number of times the task was run
@@ -824,6 +881,9 @@ definitions:
       callRoot:
         type: string
         description: If from a task, the root directory of the task or shard execution
+      operationId:
+        type: string
+        description: Operation Id for Google Pipelines API
       jobId:
         type: string
         description: >

Diff for: servers/cromwell/jobs/controllers/jobs_controller.py

@@ -24,6 +24,7 @@
 from jobs.models.execution_event import ExecutionEvent
 from jobs.models.individual_attempt import IndividualAttempt
 from jobs.models.job_attempts_response import JobAttemptsResponse
+from jobs.models.job_operation_response import JobOperationResponse
 from jobs.controllers.utils import job_statuses
 from jobs.controllers.utils import task_statuses
 
@@ -286,6 +287,7 @@ def format_task(task_name, task_metadata):
         return_code=latest_attempt.get('returnCode'),
         attempts=latest_attempt.get('attempt'),
         call_root=latest_attempt.get('callRoot'),
+        operation_id=latest_attempt.get('jobId'),
         job_id=latest_attempt.get('subWorkflowId'),
         shards=None,
         call_cached=call_cached,
@@ -301,6 +303,7 @@ def format_task_failure(task_name, metadata):
                           stdout=metadata.get('stdout'),
                           stderr=metadata.get('stderr'),
                           call_root=metadata.get('callRoot'),
+                          operation_id=metadata.get('jobId'),
                           job_id=metadata.get('subWorkflowId'))
 
 
@@ -346,6 +349,7 @@ def format_scattered_task(task_name, task_metadata):
                       stdout=shard.get('stdout'),
                       stderr=shard.get('stderr'),
                       call_root=shard.get('callRoot'),
+                      operation_id=shard.get('jobId'),
                       attempts=shard.get('attempt'),
                       failure_messages=failure_messages,
                       job_id=shard.get('subWorkflowId')))
@@ -537,6 +541,36 @@ def handle_error(response):
     response.raise_for_status()
 
 
+@requires_auth
+def get_operation_details(job, operation, **kwargs):
+    """
+    Query for operation details from Google Pipelines API
+
+    :param job: Job ID
+    :type id: str
+
+    :param operation_id: Operation ID
+    :type id: str
+
+    :rtype: str
+    """
+
+    capabilities = current_app.config['capabilities']
+    if hasattr(capabilities, 'authentication') and hasattr(
+            capabilities.authentication,
+            'outside_auth') and capabilities.authentication.outside_auth:
+        url = '{cromwell_url}/{id}/backend/metadata/{backendId}'.format(
+            cromwell_url=_get_base_url(), id=job, backendId=operation)
+        response = requests.get(url,
+                                auth=kwargs.get('auth'),
+                                headers=kwargs.get('auth_headers'))
+
+    if response.status_code != 200:
+        handle_error(response)
+
+    return JobOperationResponse(id=job, details=response.text)
+
+
 def _get_execution_events(events):
     execution_events = None
     if events:
@@ -604,6 +638,7 @@ def _convert_to_attempt(item):
         stdout=item.get('stdout'),
         stderr=item.get('stderr'),
         call_root=item.get('callRoot'),
+        operation_id=item.get('jobId'),
         inputs=item.get('inputs'),
         outputs=item.get('outputs'),
         start=_parse_datetime(item.get('start')),

Diff for: servers/cromwell/jobs/test/test_auth_utils.py

@@ -15,7 +15,8 @@ def test_token_auth_returns_200(self, mock_request):
             'cromwell_url': self.base_url,
             'cromwell_user': '',
             'cromwell_password': '',
-            'use_caas': True
+            'use_caas': True,
+            'capabilities': {}
         })
 
         def _request_callback(request, context):
@@ -38,7 +39,8 @@ def test_basic_auth_returns_200(self, mock_request):
             'cromwell_url': self.base_url,
             'cromwell_user': 'user',
             'cromwell_password': 'password',
-            'use_caas': False
+            'use_caas': False,
+            'capabilities': {}
         })
 
         def _request_callback(request, context):
@@ -59,7 +61,8 @@ def test_no_auth_with_caas_returns_401(self):
             'cromwell_url': self.base_url,
             'cromwell_user': '',
             'cromwell_password': '',
-            'use_caas': True
+            'use_caas': True,
+            'capabilities': {}
         })
         response = self.client.open('/jobs/query',
                                     method='POST',

Diff for: servers/cromwell/jobs/test/test_jobs_controller.py

@@ -27,7 +27,8 @@ def setUp(self):
             'cromwell_url': self.base_url,
             'cromwell_user': 'user',
             'cromwell_password': 'password',
-            'use_caas': False
+            'use_caas': False,
+            'capabilities': {}
         })
 
     @requests_mock.mock()

Diff for: servers/dsub/jobs/controllers/jobs_controller.py

@@ -207,6 +207,14 @@ def get_task_attempts(id, task, **kwargs):
     raise NotImplementedError('Tasks not supported by dsub.')
 
 
+def get_operation_details(job, operation, **kwargs):
+    """
+    Placeholder for query for operation details from Google Pipelines API
+
+    """
+    raise NotImplementedError('Operation details not supported by dsub.')
+
+
 def _handle_http_error(error, proj_id):
     # TODO(https://github.com/googlegenomics/dsub/issues/79): Push this
     # provider-specific error translation down into dstat.