--- /dev/null
+# ============LICENSE_START====================================================
+# org.onap.dcae
+# =============================================================================
+# Copyright (c) 2019 AT&T Intellectual Property. All rights reserved.
+# =============================================================================
+# Licensed 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.
+# ============LICENSE_END======================================================
+
+import json
+import os
+from pkg_resources import resource_string
+import shutil
+import traceback
+try:
+ from urllib.parse import unquote_plus
+ from socketserver import ThreadingMixIn
+ from http.server import BaseHTTPRequestHandler, HTTPServer
+except ImportError:
+ from urllib import unquote_plus
+ from SocketServer import ThreadingMixIn
+ from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
+
+import requests
+
+from aoconversion import convert
+
+
+def _derefconfig(value):
+ if value.startswith('@'):
+ with open(value[1:], 'r') as f:
+ return f.readline().strip()
+ return value
+
+
+class Config(object):
+ """
+ Configuration parameters as attributes, make sure the required ones are there,
+ populate defaults.
+ """
+ def __init__(self, dcaeurl, dcaeuser, onboardingurl, onboardinguser, onboardingpass, certfile, dockerregistry, dockeruser, dockerpass, acumosurl=None, interval=900, dockerhost='unix:///var/run/docker.sock', tmpdir='/var/tmp/aoadapter', certverify=True, catalogs=None, port=None, **extras):
+ self.dcaeurl = dcaeurl
+ self.dcaeuser = dcaeuser
+
+ def x(fmt, *args, **kwargs):
+ return onboardingurl + fmt.format(*args, **kwargs)
+ self.oburl = x
+ self._onboardingpass = onboardingpass
+ self._onboardinguser = onboardinguser
+ self.acumosurl = acumosurl
+ self.certfile = certfile
+ self.certverify = certverify
+ self.dockerhost = dockerhost
+ self.dockerregistry = dockerregistry
+ self.dockeruser = dockeruser
+ self._dockerpass = dockerpass
+ self.interval = interval
+ self.tmpdir = tmpdir
+ if catalogs is not None and type(catalogs) is not list:
+ catalogs = [catalogs]
+ self.catalogs = catalogs
+ self.port = port
+
+ def obauth(self):
+ return (self._onboardinguser, _derefconfig(self._onboardingpass))
+
+ def dockerpass(self):
+ return _derefconfig(self._dockerpass)
+
+
+class _AcumosAccess(object):
+ def __init__(self, config, url):
+ self.cert = config.certfile
+ self.verify = config.certverify
+ self.url = url.strip().rstrip('/')
+
+ def artgetter(self, xrev, matcher):
+ for art in xrev['artifacts']:
+ if matcher(art):
+ def ret():
+ nurl = '{}/artifacts/{}/content'.format(self.url, art['artifactId'])
+ resp = requests.get(nurl, stream=True, cert=self.cert, verify=self.verify)
+ if resp.status_code == 500:
+ resp = requests.get(nurl, stream=True, cert=self.cert, verify=self.verify)
+ resp.raise_for_status()
+ return resp
+ return ret
+ return None
+
+ def jsonget(self, format, *args, **kwargs):
+ nurl = self.url + format.format(*args, **kwargs)
+ resp = requests.get(nurl, cert=self.cert, verify=self.verify)
+ if resp.status_code == 500:
+ resp = requests.get(nurl, cert=self.cert, verify=self.verify)
+ resp.raise_for_status()
+ return resp.json()['content']
+
+
+def _x_proto_matcher(art):
+ """ Is this artifact the x.proto file? """
+ return art['name'].endswith('.proto')
+
+
+def _x_zip_matcher(art):
+ """ Is this artifact the x.zip file? """
+ return art['name'].endswith('.zip')
+
+
+def _md_json_matcher(art):
+ """ Is this artifact the metadata.json file? """
+ return art['name'].endswith('.json')
+
+
+def _walk(config):
+ """
+ Walk the Federation E5 interface of an Acumos instance
+ """
+ url = config.acumosurl
+ callback = _makecallback(config)
+ catalogs = config.catalogs
+ aa = _AcumosAccess(config, url)
+ for catalog in aa.jsonget('/catalogs'):
+ if catalogs is not None and catalog['catalogId'] not in catalogs and catalog['name'] not in catalogs:
+ continue
+ for solution in aa.jsonget('/solutions?catalogId={}', catalog['catalogId']):
+ for revision in aa.jsonget('/solutions/{}/revisions', solution['solutionId']):
+ onboard(aa, callback, solution, revision['revisionId'])
+
+
+def onboard(aa, callback, solution, revid):
+ xrev = aa.jsonget('/solutions/{}/revisions/{}', solution['solutionId'], revid)
+ 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))
+
+
+def _pullfile(source, dest):
+ with open(dest, 'wb') as f:
+ for chunk in source().iter_content(65536):
+ f.write(chunk)
+
+
+_loadedformats = set()
+_loadedcomponents = set()
+
+
+def scan(config):
+ _walk(config)
+
+
+def _makecallback(config):
+ workdir = config.tmpdir
+ obauth = config.obauth()
+ oburl = config.oburl
+
+ def callback(model_name, model_version, model_last_updated, rating, proto_getter, zip_getter, metadata_getter):
+ model_name = model_name.lower()
+ model_version = model_version.lower()
+ compid = (model_name, model_version)
+ if compid in _loadedcomponents:
+ print('Skipping component {}: already analyzed'.format(compid))
+ return
+ if proto_getter is None or zip_getter is None or metadata_getter is None:
+ print('Skipping component {}: does not have required artifacts'.format(compid))
+ _loadedcomponents.add(compid)
+ return
+ modeldir = '{}/{}'.format(workdir, model_name)
+ shutil.rmtree(modeldir, True)
+ os.makedirs(modeldir)
+ try:
+ _pullfile(proto_getter, '{}/model.proto'.format(modeldir))
+ _pullfile(zip_getter, '{}/model.zip'.format(modeldir))
+ _pullfile(metadata_getter, '{}/metadata.json'.format(modeldir))
+ except Exception:
+ print('Skipping component {}: artifact access error'.format(compid))
+ _loadedcomponents.add(compid)
+ return
+ try:
+ docker_uri, data_formats, spec = convert.gen_dcae_artifacts_for_model(config, model_name, model_version)
+ shutil.rmtree(modeldir)
+ except Exception:
+ print('Error analyzing artifacts for {}'.format(compid))
+ traceback.print_exc()
+ return
+ for data_format in data_formats:
+ fmtid = (data_format['self']['name'], data_format['self']['version'])
+ if fmtid in _loadedformats:
+ print('Skipping data format {}: already analyzed'.format(fmtid))
+ continue
+ try:
+ resp = requests.post(oburl('/dataformats'), json={'owner': config.dcaeuser, 'spec': data_format}, auth=obauth)
+ if resp.status_code == 409:
+ print('Skipping data format {}: previously loaded'.format(fmtid))
+ _loadedformats.add(fmtid)
+ continue
+ resp.raise_for_status()
+ requests.patch(resp.json()['dataFormatUrl'], json={'owner': config.dcaeuser, 'status': 'published'}, auth=obauth).raise_for_status()
+ print('Loaded data format {}'.format(fmtid))
+ _loadedformats.add(fmtid)
+ except Exception:
+ print('Error loading data format {}'.format(fmtid))
+ traceback.print_exc()
+ raise
+ try:
+ resp = requests.post(oburl('/components'), json={'owner': config.dcaeuser, 'spec': spec}, auth=obauth)
+ if resp.status_code == 409:
+ print('Skipping component {}: previously loaded'.format(compid))
+ _loadedcomponents.add(compid)
+ return
+ resp.raise_for_status()
+ requests.patch(resp.json()['componentUrl'], json={'owner': config.dcaeuser, 'status': 'published'}, auth=obauth).raise_for_status()
+ print('Loaded component {}'.format(compid))
+ _loadedcomponents.add(compid)
+ except Exception:
+ print('Error loading component {}'.format(compid))
+ traceback.print_exc()
+ raise
+ return callback
+
+
+class ThreadedHTTPServer(ThreadingMixIn, HTTPServer):
+ protocol_version = "HTTP/1.1"
+
+
+class Apihandler(BaseHTTPRequestHandler):
+ def doqp(self):
+ self.qparams = {}
+ if not self.path or '?' not in self.path:
+ return
+ self.path, qp = self.path.split('?', 1)
+ for x in qp.split('&'):
+ k, v = x.split('=', 1)
+ self.qparams[unquote_plus(k)] = unquote_plus(v)
+
+ def replyjson(self, body, ctype='application/json'):
+ self.replyraw(json.dumps(body).encode('utf-8'), ctype)
+
+ def replyraw(self, data, ctype):
+ self.send_response(200)
+ self.send_header('Content-Type', ctype)
+ self.send_header('Content-Length', len(data))
+ self.end_headers()
+ self.wfile.write(data)
+
+ def do_GET(self):
+ self.doqp()
+ if self.path == '/' or self.path == '/index.html':
+ self.replyraw(self.server.index, 'text/html')
+ return
+ if 'acumos' not in self.qparams:
+ self.send_error(400)
+ return
+ aa = _AcumosAccess(self.server.config, self.qparams['acumos'])
+ if self.path == '/listCatalogs.js':
+ self.replyjson([{'name': x['name'], 'id': x['catalogId']} for x in aa.jsonget('/catalogs')])
+ return
+ if self.path == '/listSolutions.js':
+ if 'catalogId' not in self.qparams:
+ self.send_error(400)
+ return
+ self.replyjson([{'name': x['name'], 'id': x['solutionId']} for x in aa.jsonget('/solutions?catalogId={}', self.qparams['catalogId'])])
+ return
+ if self.path == '/listRevisions.js':
+ if 'solutionId' not in self.qparams:
+ self.send_error(400)
+ return
+ self.replyjson([{'name': x['version'], 'id': x['revisionId']} for x in aa.jsonget('/solutions/{}/revisions', self.qparams['solutionId'])])
+ return
+ self.send_error(404)
+
+ def do_POST(self):
+ self.doqp()
+ if self.path == '/onboard.js':
+ if 'acumos' not in self.qparams:
+ self.send_error(400)
+ return
+ aa = _AcumosAccess(self.server.config, self.qparams['acumos'])
+ callback = self.server.callback
+ if 'catalogId' not in self.qparams:
+ for catalog in aa.jsonget('/catalogs'):
+ for solution in aa.jsonget('/solutions?catalogId={}', catalog['catalogId']):
+ for revision in aa.jsonget('/solutions/{}/revisions', solution['solutionId']):
+ onboard(aa, callback, solution, revision['revisionId'])
+ elif 'solutionId' not in self.qparams:
+ for solution in aa.jsonget('/solutions?catalogId={}', self.qparams['catalogId']):
+ for revision in aa.jsonget('/solutions/{}/revisions', solution['solutionId']):
+ onboard(aa, callback, solution, revision['revisionId'])
+ elif 'revisionId' not in self.qparams:
+ solution = aa.jsonget('/solutions/{}', self.qparams['solutionId'])
+ for revision in aa.jsonget('/solutions/{}/revisions', solution['solutionId']):
+ onboard(aa, callback, solution, revision['revisionId'])
+ else:
+ solution = aa.jsonget('/solutions/{}', self.qparams['solutionId'])
+ onboard(aa, callback, solution, self.qparams['revisionId'])
+ self.replyraw('OK', 'text/plain')
+ return
+ self.send_error(400)
+
+
+def serve(config):
+ server = ThreadedHTTPServer(('', config.port), Apihandler)
+ server.config = config
+ server.callback = _makecallback(config)
+ server.index = resource_string(__name__, 'index.html')
+ server.serve_forever()
Code Review / dcaegen2 / platform.git / blobdiff