adapter/acumos/aoconversion/scanner.py

   1 # ============LICENSE_START====================================================
   2 # org.onap.dcae
   3 # =============================================================================
   4 # Copyright (c) 2019-2020 AT&T Intellectual Property. All rights reserved.
   5 # =============================================================================
   6 # Licensed under the Apache License, Version 2.0 (the "License");
   7 # you may not use this file except in compliance with the License.
   8 # You may obtain a copy of the License at
   9 #
  10 #      http://www.apache.org/licenses/LICENSE-2.0
  11 #
  12 # Unless required by applicable law or agreed to in writing, software
  13 # distributed under the License is distributed on an "AS IS" BASIS,
  14 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15 # See the License for the specific language governing permissions and
  16 # limitations under the License.
  17 # ============LICENSE_END======================================================
  18
  19 import json
  20 import os
  21 from pkg_resources import resource_string
  22 import shutil
  23 import traceback
  24 try:
  25     from urllib.parse import unquote_plus
  26     from socketserver import ThreadingMixIn
  27     from http.server import BaseHTTPRequestHandler, HTTPServer
  28 except ImportError:
  29     from urllib import unquote_plus
  30     from SocketServer import ThreadingMixIn
  31     from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
  32
  33 import requests
  34
  35 from aoconversion import convert
  36
  37
  38 def _derefconfig(value):
  39     if value.startswith('@'):
  40         with open(value[1:], 'r') as f:
  41             return f.readline().strip()
  42     return value
  43
  44
  45 class Config(object):
  46     """
  47     Configuration parameters as attributes, make sure the required ones are there,
  48     populate defaults.
  49     """
  50     def __init__(self, dcaeuser, onboardingurl, onboardinguser, onboardingpass, certfile, dockerregistry, dockeruser, dockerpass, acumosurl=None, interval=900, dockerhost='tcp://localhost:2375', tmpdir='/var/tmp/aoadapter', certverify=True, catalogs=None, port=None, **extras):
  51         self.dcaeuser = dcaeuser
  52
  53         def x(fmt, *args, **kwargs):
  54             return onboardingurl + fmt.format(*args, **kwargs)
  55         self.oburl = x
  56         self._onboardingpass = onboardingpass
  57         self._onboardinguser = onboardinguser
  58         self.acumosurl = acumosurl
  59         self.certfile = certfile
  60         self.certverify = certverify
  61         self.dockerhost = dockerhost
  62         self.dockerregistry = dockerregistry
  63         self.dockeruser = dockeruser
  64         self._dockerpass = dockerpass
  65         self.interval = interval
  66         self.tmpdir = tmpdir
  67         if catalogs is not None and type(catalogs) is not list:
  68             catalogs = [catalogs]
  69         self.catalogs = catalogs
  70         self.port = port
  71
  72     def obauth(self):
  73         return (self._onboardinguser, _derefconfig(self._onboardingpass))
  74
  75     def dockerpass(self):
  76         return _derefconfig(self._dockerpass)
  77
  78
  79 class _AcumosAccess(object):
  80     def __init__(self, config, url):
  81         self.cert = config.certfile
  82         self.verify = config.certverify
  83         self.url = url.strip().rstrip('/')
  84
  85     def artgetter(self, xrev, matcher):
  86         for art in xrev['artifacts']:
  87             if matcher(art):
  88                 def ret():
  89                     nurl = '{}/artifacts/{}/content'.format(self.url, art['artifactId'])
  90                     resp = requests.get(nurl, stream=True, cert=self.cert, verify=self.verify)
  91                     if resp.status_code == 500:
  92                         resp = requests.get(nurl, stream=True, cert=self.cert, verify=self.verify)
  93                     resp.raise_for_status()
  94                     return resp
  95                 return ret
  96         return None
  97
  98     def jsonget(self, format, *args, **kwargs):
  99         nurl = self.url + format.format(*args, **kwargs)
 100         resp = requests.get(nurl, cert=self.cert, verify=self.verify)
 101         if resp.status_code == 500:
 102             resp = requests.get(nurl, cert=self.cert, verify=self.verify)
 103         resp.raise_for_status()
 104         return resp.json()['content']
 105
 106
 107 def _x_proto_matcher(art):
 108     """ Is this artifact the x.proto file? """
 109     return art['name'].endswith('.proto')
 110
 111
 112 def _x_zip_matcher(art):
 113     """ Is this artifact the x.zip file? """
 114     return art['name'].endswith('.zip')
 115
 116
 117 def _md_json_matcher(art):
 118     """ Is this artifact the metadata.json file? """
 119     return art['name'].endswith('.json')
 120
 121
 122 def _walk(config):
 123     """
 124     Walk the Federation E5 interface of an Acumos instance
 125     """
 126     url = config.acumosurl
 127     callback = _makecallback(config)
 128     catalogs = config.catalogs
 129     aa = _AcumosAccess(config, url)
 130     for catalog in aa.jsonget('/catalogs'):
 131         if catalogs is not None and catalog['catalogId'] not in catalogs and catalog['name'] not in catalogs:
 132             continue
 133         for solution in aa.jsonget('/solutions?catalogId={}', catalog['catalogId']):
 134             for revision in aa.jsonget('/solutions/{}/revisions', solution['solutionId']):
 135                 onboard(aa, callback, solution, revision['revisionId'])
 136
 137
 138 def onboard(aa, callback, solution, revid):
 139     xrev = aa.jsonget('/solutions/{}/revisions/{}', solution['solutionId'], revid)
 140     callback(model_name=solution['name'], model_version=xrev['version'], model_last_updated=xrev['modified'], rating=solution['ratingAverageTenths'] / 10.0, proto_getter=aa.artgetter(xrev, _x_proto_matcher), zip_getter=aa.artgetter(xrev, _x_zip_matcher), metadata_getter=aa.artgetter(xrev, _md_json_matcher))
 141
 142
 143 def _pullfile(source, dest):
 144     with open(dest, 'wb') as f:
 145         for chunk in source().iter_content(65536):
 146             f.write(chunk)
 147
 148
 149 _loadedformats = set()
 150 _loadedcomponents = set()
 151
 152
 153 def scan(config):
 154     _walk(config)
 155
 156
 157 def _makecallback(config):
 158     workdir = config.tmpdir
 159     obauth = config.obauth()
 160     oburl = config.oburl
 161
 162     def callback(model_name, model_version, model_last_updated, rating, proto_getter, zip_getter, metadata_getter):
 163         model_name = model_name.lower()
 164         model_version = model_version.lower()
 165         compid = (model_name, model_version)
 166         if compid in _loadedcomponents:
 167             print('Skipping component {}: already analyzed'.format(compid))
 168             return
 169         if proto_getter is None or zip_getter is None or metadata_getter is None:
 170             print('Skipping component {}: does not have required artifacts'.format(compid))
 171             _loadedcomponents.add(compid)
 172             return
 173         modeldir = '{}/{}'.format(workdir, model_name)
 174         shutil.rmtree(modeldir, True)
 175         os.makedirs(modeldir)
 176         try:
 177             _pullfile(proto_getter, '{}/model.proto'.format(modeldir))
 178             _pullfile(zip_getter, '{}/model.zip'.format(modeldir))
 179             _pullfile(metadata_getter, '{}/metadata.json'.format(modeldir))
 180         except Exception:
 181             print('Skipping component {}: artifact access error'.format(compid))
 182             _loadedcomponents.add(compid)
 183             return
 184         try:
 185             docker_uri, data_formats, spec = convert.gen_dcae_artifacts_for_model(config, model_name, model_version)
 186             shutil.rmtree(modeldir)
 187         except Exception:
 188             print('Error analyzing artifacts for {}'.format(compid))
 189             traceback.print_exc()
 190             return
 191         for data_format in data_formats:
 192             fmtid = (data_format['self']['name'], data_format['self']['version'])
 193             if fmtid in _loadedformats:
 194                 print('Skipping data format {}: already analyzed'.format(fmtid))
 195                 continue
 196             try:
 197                 resp = requests.post(oburl('/dataformats'), json={'owner': config.dcaeuser, 'spec': data_format}, auth=obauth)
 198                 if resp.status_code == 409:
 199                     print('Skipping data format {}: previously loaded'.format(fmtid))
 200                     _loadedformats.add(fmtid)
 201                     continue
 202                 resp.raise_for_status()
 203                 requests.patch(resp.json()['dataFormatUrl'], json={'owner': config.dcaeuser, 'status': 'published'}, auth=obauth).raise_for_status()
 204                 print('Loaded data format {}'.format(fmtid))
 205                 _loadedformats.add(fmtid)
 206             except Exception:
 207                 print('Error loading data format {}'.format(fmtid))
 208                 traceback.print_exc()
 209                 raise
 210         try:
 211             resp = requests.post(oburl('/components'), json={'owner': config.dcaeuser, 'spec': spec}, auth=obauth)
 212             if resp.status_code == 409:
 213                 print('Skipping component {}: previously loaded'.format(compid))
 214                 _loadedcomponents.add(compid)
 215                 return
 216             resp.raise_for_status()
 217             requests.patch(resp.json()['componentUrl'], json={'owner': config.dcaeuser, 'status': 'published'}, auth=obauth).raise_for_status()
 218             print('Loaded component {}'.format(compid))
 219             _loadedcomponents.add(compid)
 220         except Exception:
 221             print('Error loading component {}'.format(compid))
 222             traceback.print_exc()
 223             raise
 224     return callback
 225
 226
 227 class ThreadedHTTPServer(ThreadingMixIn, HTTPServer):
 228     protocol_version = "HTTP/1.1"
 229
 230
 231 class Apihandler(BaseHTTPRequestHandler):
 232     def doqp(self):
 233         self.qparams = {}
 234         if not self.path or '?' not in self.path:
 235             return
 236         self.path, qp = self.path.split('?', 1)
 237         for x in qp.split('&'):
 238             k, v = x.split('=', 1)
 239             self.qparams[unquote_plus(k)] = unquote_plus(v)
 240
 241     def replyjson(self, body, ctype='application/json'):
 242         self.replyraw(json.dumps(body).encode('utf-8'), ctype)
 243
 244     def replyraw(self, data, ctype):
 245         self.send_response(200)
 246         self.send_header('Content-Type', ctype)
 247         self.send_header('Content-Length', len(data))
 248         self.end_headers()
 249         self.wfile.write(data)
 250
 251     def do_GET(self):
 252         self.doqp()
 253         if self.path == '/' or self.path == '/index.html' or self.path == '/acumos-adapter/' or self.path == '/acumos-adapter/index.html':
 254             self.replyraw(self.server.index, 'text/html')
 255             return
 256         if 'acumos' not in self.qparams:
 257             self.send_error(400)
 258             return
 259         aa = _AcumosAccess(self.server.config, self.qparams['acumos'])
 260         if self.path == '/acumos-adapter/listCatalogs.js':
 261             self.replyjson([{'name': x['name'], 'id': x['catalogId']} for x in aa.jsonget('/catalogs')])
 262             return
 263         if self.path == '/acumos-adapter/listSolutions.js':
 264             if 'catalogId' not in self.qparams:
 265                 self.send_error(400)
 266                 return
 267             self.replyjson([{'name': x['name'], 'id': x['solutionId']} for x in aa.jsonget('/solutions?catalogId={}', self.qparams['catalogId'])])
 268             return
 269         if self.path == '/acumos-adapter/listRevisions.js':
 270             if 'solutionId' not in self.qparams:
 271                 self.send_error(400)
 272                 return
 273             self.replyjson([{'name': x['version'], 'id': x['revisionId']} for x in aa.jsonget('/solutions/{}/revisions', self.qparams['solutionId'])])
 274             return
 275         self.send_error(404)
 276
 277     def do_POST(self):
 278         self.doqp()
 279         if self.path == '/acumos-adapter/onboard.js':
 280             if 'acumos' not in self.qparams:
 281                 self.send_error(400)
 282                 return
 283             aa = _AcumosAccess(self.server.config, self.qparams['acumos'])
 284             callback = self.server.callback
 285             if 'catalogId' not in self.qparams:
 286                 for catalog in aa.jsonget('/catalogs'):
 287                     for solution in aa.jsonget('/solutions?catalogId={}', catalog['catalogId']):
 288                         for revision in aa.jsonget('/solutions/{}/revisions', solution['solutionId']):
 289                             onboard(aa, callback, solution, revision['revisionId'])
 290             elif 'solutionId' not in self.qparams:
 291                 for solution in aa.jsonget('/solutions?catalogId={}', self.qparams['catalogId']):
 292                     for revision in aa.jsonget('/solutions/{}/revisions', solution['solutionId']):
 293                         onboard(aa, callback, solution, revision['revisionId'])
 294             elif 'revisionId' not in self.qparams:
 295                 solution = aa.jsonget('/solutions/{}', self.qparams['solutionId'])
 296                 for revision in aa.jsonget('/solutions/{}/revisions', solution['solutionId']):
 297                     onboard(aa, callback, solution, revision['revisionId'])
 298             else:
 299                 solution = aa.jsonget('/solutions/{}', self.qparams['solutionId'])
 300                 onboard(aa, callback, solution, self.qparams['revisionId'])
 301             self.replyraw('OK'.encode('utf-8'), 'text/plain')
 302             return
 303         self.send_error(400)
 304
 305
 306 def serve(config):
 307     server = ThreadedHTTPServer(('', config.port), Apihandler)
 308     server.config = config
 309     server.callback = _makecallback(config)
 310     server.index = resource_string(__name__, 'index.html')
 311     server.serve_forever()