Better error messaging on missing gcloud

Also don't retry certain permanent auth errors.

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

Author: robertwb

google/cloud/dataflow/internal/auth.py

@@ -25,6 +25,7 @@
 from oauth2client.client import OAuth2Credentials
 
 from google.cloud.dataflow.utils import processes
+from google.cloud.dataflow.utils import retry
 from google.cloud.dataflow.utils.options import GoogleCloudOptions
 from google.cloud.dataflow.utils.options import PipelineOptions
 
@@ -53,6 +54,10 @@ def set_running_in_gce(worker_executing_project):
   executing_project = worker_executing_project
 
 
+class AuthenticationException(retry.PermanentException):
+  pass
+
+
 class GCEMetadataCredentials(OAuth2Credentials):
   """Credential object initialized using access token from GCE VM metadata."""
 
@@ -98,8 +103,8 @@ def _refresh(self, http_request):
       gcloud_process = processes.Popen(
           ['gcloud', 'auth', 'print-access-token'], stdout=processes.PIPE)
     except OSError as exn:
-      logging.error('The gcloud tool was not found.')
-      raise exn
+      logging.error('The gcloud tool was not found.', exc_info=True)
+      raise AuthenticationException('The gcloud tool was not found: %s' % exn)
     output, _ = gcloud_process.communicate()
     self.access_token = output.strip()
 
@@ -120,9 +125,11 @@ def get_service_credentials():
         sys.argv).view_as(GoogleCloudOptions)
     if google_cloud_options.service_account_name:
       if not google_cloud_options.service_account_key_file:
-        raise Exception('key file not provided for service account.')
+        raise AuthenticationException(
+            'key file not provided for service account.')
       if not os.path.exists(google_cloud_options.service_account_key_file):
-        raise Exception('Specified service account key file does not exist.')
+        raise AuthenticationException(
+            'Specified service account key file does not exist.')
       client_scopes = [
           'https://www.googleapis.com/auth/bigquery',
           'https://www.googleapis.com/auth/cloud-platform',

google/cloud/dataflow/utils/retry.py

@@ -29,6 +29,11 @@
 from apitools.base.py.exceptions import HttpError
 
 
+class PermanentException(Exception):
+  """Base class for exceptions that should not be retried."""
+  pass
+
+
 class FuzzedExponentialIntervals(object):
   """Iterable for intervals that are exponentially spaced, with fuzzing.
 
@@ -75,6 +80,8 @@ def retry_on_server_errors_filter(exception):
       return True
     else:
       return False
+  elif isinstance(exception, PermanentException):
+    return False
   else:
     # We may get here for non HttpErrors such as socket timeouts, SSL
     # exceptions, etc.