| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414 |
- #!/usr/bin/env python
- # -- coding: utf-8 --
- # Licensed to Cloudera, Inc. under one
- # or more contributor license agreements. See the NOTICE file
- # distributed with this work for additional information
- # regarding copyright ownership. Cloudera, Inc. licenses this file
- # to you under the Apache License, Version 2.0 (the
- # "License"); you may not use this file except in compliance
- # with the License. You may obtain a copy of the License at
- #
- # http://www.apache.org/licenses/LICENSE-2.0
- #
- # Unless required by applicable law or agreed to in writing, software
- # distributed under the License is distributed on an "AS IS" BASIS,
- # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- # See the License for the specific language governing permissions and
- # limitations under the License.
- from builtins import object
- import json
- import logging
- import os
- import time
- import uuid
- from tempfile import NamedTemporaryFile
- from django.core.cache import cache
- from django.utils.functional import wraps
- from django.utils.translation import ugettext as _
- from desktop.auth.backend import is_admin
- from desktop.lib.exceptions_renderable import PopupException
- from desktop.lib import export_csvxls
- from desktop.lib.i18n import smart_unicode
- from desktop.lib.rest.http_client import RestException
- from libsentry.sentry_site import get_hive_sentry_provider
- from libsentry.privilege_checker import get_checker, MissingSentryPrivilegeException
- from metadata.conf import OPTIMIZER, get_optimizer_url
- LOG = logging.getLogger(__name__)
- _JSON_CONTENT_TYPE = 'application/json'
- OPTIMIZER_TENANT_ID_CACHE_KEY = 'navopt-tenant-id'
- try:
- from navoptapi.api_lib import ApiLib
- except Exception as e:
- LOG.warn('NavOpt module is not installed: %s' % e)
- class NavOptException(Exception):
- def __init__(self, message=None):
- self.message = message or _('No error message, please check the logs.')
- def __str__(self):
- return str(self.message)
- def __unicode__(self):
- return smart_unicode(self.message)
- def check_privileges(view_func):
- def decorate(*args, **kwargs):
- if OPTIMIZER.APPLY_SENTRY_PERMISSIONS.get():
- checker = get_checker(user=args[0].user)
- action = 'SELECT'
- objects = []
- if kwargs.get('db_tables'):
- for db_table in kwargs['db_tables']:
- objects.append({'server': get_hive_sentry_provider(), 'db': _get_table_name(db_table)['database'], 'table': _get_table_name(db_table)['table']})
- else:
- objects = [{'server': get_hive_sentry_provider()}]
- if kwargs.get('database_name'):
- objects[0]['db'] = kwargs['database_name']
- if kwargs.get('table_name'):
- objects[0]['table'] = kwargs['table_name']
- filtered = list(checker.filter_objects(objects, action))
- if len(filtered) != len(objects):
- raise MissingSentryPrivilegeException({'pre_filtering': objects, 'post_filtering': filtered, 'diff': len(objects) - len(filtered)})
- return view_func(*args, **kwargs)
- return wraps(view_func)(decorate)
- class OptimizerApi(object):
- def __init__(self, user, api_url=None, auth_key=None, auth_key_secret=None, tenant_id=None):
- self.user = user
- self._api_url = (api_url or get_optimizer_url()).strip('/')
- self._auth_key = auth_key if auth_key else OPTIMIZER.AUTH_KEY_ID.get()
- self._auth_key_secret = auth_key_secret if auth_key_secret else (OPTIMIZER.AUTH_KEY_SECRET.get() and OPTIMIZER.AUTH_KEY_SECRET.get().replace('\\n', '\n'))
- self._api = ApiLib("navopt", self._api_url, self._auth_key, self._auth_key_secret)
- self._tenant_id = tenant_id if tenant_id else _get_tenant_id(self) # Aka "workload"
- def _call(self, *kwargs):
- start_time = time.time()
- resp = self._api.call_api(*kwargs)
- data = resp.json()
- if resp.headers.get('x-altus-request-id'):
- LOG.info('%s %s in %dms: %s' % (self.user, resp.headers['x-altus-request-id'], (time.time() - start_time) * 1000, kwargs))
- if data.get('code') == 'UNKNOWN':
- raise NavOptException(data.get('message'))
- elif data.get('errorMsg'):
- raise NavOptException(data.get('errorMsg'))
- else:
- return data
- def get_tenant(self, cluster_id='default'):
- return self._call('getTenant', {'clusterId' : cluster_id})
- def upload(self, data, data_type='queries', source_platform='generic', workload_id=None):
- if data_type in ('table_stats', 'cols_stats'):
- data_suffix = '.json'
- if data_type == 'table_stats':
- extra_parameters = {'fileType': 'TABLE_STATS'}
- else:
- extra_parameters = {'fileType': 'COLUMN_STATS'}
- else:
- data_suffix = '.csv'
- extra_parameters = {
- 'fileType': 'QUERY',
- 'colDelim': ',',
- 'rowDelim': '\n',
- "headerFields": [
- {"count": 0, "name": "SQL_ID", "coltype": "SQL_ID", "use": True, "tag": ""},
- {"count": 0, "name": "ELAPSED_TIME", "coltype": "NONE", "use": True, "tag": ""},
- {"count": 0, "name": "SQL_FULLTEXT", "coltype": "SQL_QUERY", "use": True, "tag": ""},
- {"count": 0, "name": "DATABASE", "coltype": "NONE", "use": True, "tag": "DATABASE"}
- ],
- }
- f_queries_path = NamedTemporaryFile(suffix=data_suffix)
- f_queries_path.close() # Reopened as real file below to work well with the command
- try:
- f_queries = open(f_queries_path.name, 'w+')
- try:
- # Queries
- if data_suffix == '.csv':
- content_generator = OptimizerQueryDataAdapter(data)
- queries_csv = export_csvxls.create_generator(content_generator, 'csv')
- for row in queries_csv:
- f_queries.write(row)
- LOG.debug(row[:1000])
- else:
- # Table, column stats
- f_queries.write(json.dumps(data))
- LOG.debug(json.dumps(data[:10]))
- finally:
- f_queries.close()
- parameters = {
- 'tenant' : self._tenant_id,
- 'fileLocation': f_queries.name,
- 'sourcePlatform': source_platform,
- }
- parameters.update(extra_parameters)
- response = self._api.call_api('upload', parameters)
- status = json.loads(response)
- status['count'] = len(data)
- return status
- except RestException as e:
- raise PopupException(e, title=_('Error while accessing Optimizer'))
- finally:
- os.remove(f_queries_path.name)
- def upload_status(self, workload_id):
- return self._call('uploadStatus', {'tenant' : self._tenant_id, 'workloadId': workload_id})
- # Sentry permissions work bottom to top.
- # @check_privileges
- def top_tables(self, workfloadId=None, database_name='default', page_size=1000, startingToken=None):
- data = self._call('getTopTables', {'tenant' : self._tenant_id, 'dbName': database_name.lower(), 'pageSize': page_size, 'startingToken': startingToken})
- if OPTIMIZER.APPLY_SENTRY_PERMISSIONS.get():
- checker = get_checker(user=self.user)
- action = 'SELECT'
- def getkey(table):
- names = _get_table_name(table['name'])
- return {'server': get_hive_sentry_provider(), 'db': names['database'], 'table': names['table']}
- data['results'] = list(checker.filter_objects(data['results'], action, key=getkey))
- return data
- @check_privileges
- def table_details(self, database_name, table_name, page_size=100, startingToken=None):
- return self._call('getTablesDetail', {'tenant' : self._tenant_id, 'dbName': database_name.lower(), 'tableName': table_name.lower(), 'pageSize': page_size, 'startingToken': startingToken})
- def query_compatibility(self, source_platform, target_platform, query, page_size=100, startingToken=None):
- return self._call('getQueryCompatible', {'tenant' : self._tenant_id, 'query': query, 'sourcePlatform': source_platform, 'targetPlatform': target_platform, 'startingToken': startingToken})
- def query_risk(self, query, source_platform, db_name, page_size=100, startingToken=None):
- response = self._call('getQueryRisk', {
- 'tenant' : self._tenant_id,
- 'query': _clean_query(query),
- 'dbName': db_name,
- 'sourcePlatform': source_platform,
- 'pageSize': page_size,
- 'startingToken': startingToken
- })
- hints = response.get(source_platform + 'Risk', {})
- if hints and hints == [{"riskTables": [], "riskAnalysis": "", "riskId": 0, "risk": "low", "riskRecommendation": ""}]:
- hints = []
- return {
- 'hints': hints,
- 'noStats': response.get('noStats', []),
- 'noDDL': response.get('noDDL', []),
- }
- def similar_queries(self, source_platform, query, page_size=100, startingToken=None):
- if is_admin(self.user):
- return self._call('getSimilarQueries', {'tenant' : self._tenant_id, 'sourcePlatform': source_platform, 'query': query, 'pageSize': page_size, 'startingToken': startingToken})
- else:
- raise PopupException(_('Call not supported'))
- @check_privileges
- def top_filters(self, db_tables=None, page_size=100, startingToken=None):
- args = {
- 'tenant' : self._tenant_id,
- 'pageSize': page_size,
- 'startingToken': startingToken
- }
- if db_tables:
- args['dbTableList'] = [db_table.lower() for db_table in db_tables]
- results = self._call('getTopFilters', args)
- if OPTIMIZER.APPLY_SENTRY_PERMISSIONS.get():
- filtered_filters = []
- for result in results['results']:
- cols = [_get_table_name(col['columnName']) for col in result["popularValues"][0]["group"]]
- if len(cols) == len(list(_secure_results(cols, self.user))):
- filtered_filters.append(result)
- results['results'] = filtered_filters
- return results
- @check_privileges
- def top_aggs(self, db_tables=None, page_size=100, startingToken=None):
- args = {
- 'tenant' : self._tenant_id,
- 'pageSize': page_size,
- 'startingToken': startingToken
- }
- if db_tables:
- args['dbTableList'] = [db_table.lower() for db_table in db_tables]
- results = self._call('getTopAggs', args)
- if OPTIMIZER.APPLY_SENTRY_PERMISSIONS.get():
- checker = get_checker(user=self.user)
- action = 'SELECT'
- def getkey(table):
- names = table['aggregateInfo'][0]
- names['server'] = get_hive_sentry_provider()
- return names
- results['results'] = list(checker.filter_objects(results['results'], action, key=getkey))
- return results
- @check_privileges
- def top_columns(self, db_tables=None, page_size=100, startingToken=None):
- args = {
- 'tenant' : self._tenant_id,
- 'pageSize': page_size,
- 'startingToken': startingToken
- }
- if db_tables:
- args['dbTableList'] = [db_table.lower() for db_table in db_tables]
- results = self._call('getTopColumns', args)
- if OPTIMIZER.APPLY_SENTRY_PERMISSIONS.get():
- for section in ['orderbyColumns', 'selectColumns', 'filterColumns', 'joinColumns', 'groupbyColumns']:
- results[section] = list(_secure_results(results[section], self.user))
- return results
- @check_privileges
- def top_joins(self, db_tables=None, page_size=100, startingToken=None):
- args = {
- 'tenant' : self._tenant_id,
- 'pageSize': page_size,
- 'startingToken': startingToken
- }
- if db_tables:
- args['dbTableList'] = [db_table.lower() for db_table in db_tables]
- results = self._call('getTopJoins', args)
- if OPTIMIZER.APPLY_SENTRY_PERMISSIONS.get():
- filtered_joins = []
- for result in results['results']:
- cols = [_get_table_name(col) for col in result["joinCols"][0]["columns"]]
- if len(cols) == len(list(_secure_results(cols, self.user))):
- filtered_joins.append(result)
- results['results'] = filtered_joins
- return results
- def top_databases(self, page_size=100, startingToken=None):
- args = {
- 'tenant' : self._tenant_id,
- 'pageSize': page_size,
- 'startingToken': startingToken
- }
- data = self._call('getTopDatabases', args)
- if OPTIMIZER.APPLY_SENTRY_PERMISSIONS.get():
- data['results'] = list(_secure_results(data['results'], self.user))
- return data
- def OptimizerQueryDataAdapter(data):
- headers = ['SQL_ID', 'ELAPSED_TIME', 'SQL_FULLTEXT', 'DATABASE']
- if data and len(data[0]) == 4:
- rows = data
- else:
- rows = ([str(uuid.uuid4()), 0.0, q, 'default'] for q in data)
- yield headers, rows
- def _get_table_name(path):
- column = None
- if path.count('.') == 1:
- database, table = path.split('.', 1)
- elif path.count('.') == 2:
- database, table, column = path.split('.', 2)
- else:
- database, table = 'default', path
- name = {'database': database, 'table': table}
- if column:
- name['column'] = column
- return name
- def _secure_results(results, user, action='SELECT'):
- if OPTIMIZER.APPLY_SENTRY_PERMISSIONS.get():
- checker = get_checker(user=user)
- def getkey(result):
- key = {'server': get_hive_sentry_provider()}
- if 'dbName' in result:
- key['db'] = result['dbName']
- elif 'database' in result:
- key['db'] = result['database']
- if 'tableName' in result:
- key['table'] = result['tableName']
- elif 'table' in result:
- key['table'] = result['table']
- if 'columnName' in result:
- key['column'] = result['columnName']
- elif 'column' in result:
- key['column'] = result['column']
- return key
- return checker.filter_objects(results, action, key=getkey)
- else:
- return results
- def _clean_query(query):
- return ' '.join([line for line in query.strip().splitlines() if not line.strip().startswith('--')])
- def _get_tenant_id(api):
- tenant_id = OPTIMIZER.TENANT_ID.get() or cache.get(OPTIMIZER_TENANT_ID_CACHE_KEY)
- if not tenant_id:
- tenant = api.get_tenant(cluster_id=OPTIMIZER.CLUSTER_ID.get())
- if tenant.get('tenant'):
- tenant_id = tenant['tenant']
- else:
- raise PopupException(_('Could not get tenant id from cluster id %s: %s') % (OPTIMIZER.CLUSTER_ID.get(), tenant))
- cache.set(OPTIMIZER_TENANT_ID_CACHE_KEY, tenant_id, 60 * 60 * 24 * 30)
- return tenant_id
|