OODT-753: Move FM and WM Python APIs into agility component by kelly, mattmann.
Project: http://git-wip-us.apache.org/repos/asf/oodt/repo Commit: http://git-wip-us.apache.org/repos/asf/oodt/commit/d329e1ab Tree: http://git-wip-us.apache.org/repos/asf/oodt/tree/d329e1ab Diff: http://git-wip-us.apache.org/repos/asf/oodt/diff/d329e1ab Branch: refs/heads/feature/zookeeper-config Commit: d329e1ab90a0aacd601859e37e3acac978d68c94 Parents: 2fc7bce Author: Chris Mattmann <[email protected]> Authored: Sun Jul 16 14:03:23 2017 -0700 Committer: Chris Mattmann <[email protected]> Committed: Sun Jul 16 14:03:23 2017 -0700 ---------------------------------------------------------------------- CHANGES.txt | 2 + agility/oodt/conf/fm.conf | 43 ++ agility/oodt/conf/setclasspath.sh | 17 + agility/oodt/fm.py | 554 +++++++++++++++++++ agility/oodt/metadata.py | 29 + agility/oodt/workflow.py | 134 +++++ filemgr/src/main/python/fm.conf | 43 -- filemgr/src/main/python/fm.py | 554 ------------------- filemgr/src/main/python/setclasspath.sh | 17 - .../main/python/xmlrpc-wrapper/oodt_metadata.py | 29 - .../xmlrpc-wrapper/oodt_workflow_xmlrpc.py | 134 ----- 11 files changed, 779 insertions(+), 777 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/oodt/blob/d329e1ab/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index 7ceb353..c4be36b 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -3,6 +3,8 @@ Apache OODT Change Log Release 1.1 - Current Development +* OODT-753 Move FM and WM Python APIs into "agility" component (kelly, mattmann) + * OODT-563 Task editing and Workflow execution (varun, mattmann) * OODT-212 Rollback capability for Workflows (sherylj via mattmann) http://git-wip-us.apache.org/repos/asf/oodt/blob/d329e1ab/agility/oodt/conf/fm.conf ---------------------------------------------------------------------- diff --git a/agility/oodt/conf/fm.conf b/agility/oodt/conf/fm.conf new file mode 100644 index 0000000..dae6438 --- /dev/null +++ b/agility/oodt/conf/fm.conf @@ -0,0 +1,43 @@ +# encoding: utf-8 +#/* +# * Licensed to the Apache Software Foundation (ASF) under one or more +# * contributor license agreements. See the NOTICE file distributed with +# * this work for additional information regarding copyright ownership. +# * The ASF 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. +# */ + +[index] +# The path gives the name of the directory where the Lucene search engine +# will build its indexes. The path refers to a directory; it will be created +# if it doesn't exist. If it does exist, it must refer to an +# existing Lucene index with correct sub-files present. So, don't make this +# directory manually, or you'll be confused by error messages! +path = @INDEX_PATH@ +pageSize = 20 + +[policies] +# The 'repo' and 'validation' are comma-separated lists of URLs. Typically, they +# contain a single "file:" style URL that names a directory with policy XML files +# within. The 'user' identifies the user database. It'll be created if it +# doesn't exist. +repo = @REPO_POLICY_URLS@ +validation = @VALIDATION_POLICY_URLS@ +user = @USER_DB_PATH@ + +[factories] +# There's no need to change anything here unless you really know what you're doing, +# and trust me, you don't. Don't take it personally. It's just a fact. +catalog = org.apache.oodt.cas.filemgr.catalog.LuceneCatalogFactory +repository = org.apache.oodt.cas.filemgr.repository.XMLRepositoryManagerFactory +datatransfer = org.apache.oodt.cas.filemgr.datatransfer.LocalDataTransferFactory +validation =org.apache.oodt.cas.filemgr.validation.XMLValidationLayerFactory http://git-wip-us.apache.org/repos/asf/oodt/blob/d329e1ab/agility/oodt/conf/setclasspath.sh ---------------------------------------------------------------------- diff --git a/agility/oodt/conf/setclasspath.sh b/agility/oodt/conf/setclasspath.sh new file mode 100755 index 0000000..1fbfc48 --- /dev/null +++ b/agility/oodt/conf/setclasspath.sh @@ -0,0 +1,17 @@ +#/* +# * Licensed to the Apache Software Foundation (ASF) under one or more +# * contributor license agreements. See the NOTICE file distributed with +# * this work for additional information regarding copyright ownership. +# * The ASF 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. +# */ +setenv CLASSPATH ${HOME}/.m2/repository/org.apache.oodt.cas-filemgr/1.7.0-dev/cas-filemgr-1.7.0-dev.jar:${HOME}/.m2/repository/xmlrpc/xmlrpc/2.1-dev/xmlrpc-2.1-dev.jar:${HOME}/.m2/repository/org.apache.oodt.cas-metadata/1.5.0/cas-metadata-1.5.0.jar:${HOME}/.m2/repository/org/apache/lucene/lucene-core/2.0.0/lucene-core-2.0.0.jar:${HOME}/.m2/repository/jug/jug-asl/2.0.0/jug-asl-2.0.0.jar http://git-wip-us.apache.org/repos/asf/oodt/blob/d329e1ab/agility/oodt/fm.py ---------------------------------------------------------------------- diff --git a/agility/oodt/fm.py b/agility/oodt/fm.py new file mode 100644 index 0000000..5eed046 --- /dev/null +++ b/agility/oodt/fm.py @@ -0,0 +1,554 @@ +#!/usr/bin/env python +# encoding: utf-8 +#/* +# * Licensed to the Apache Software Foundation (ASF) under one or more +# * contributor license agreements. See the NOTICE file distributed with +# * this work for additional information regarding copyright ownership. +# * The ASF 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. +# */ +''' +CAS Filemgr Python Server + +This is the Python server API for an authenticated Catalog and Archive System. It uses the +OODT Catalog and Archive System as its core, adding user/password-based authentication +and role-based authorization. +''' + +import sys, os, os.path +import getopt, sha, pickle + +from ConfigParser import ConfigParser +from org.apache.oodt.cas.filemgr.system.auth import SecureWebServer, Dispatcher, Result +from org.apache.oodt.cas.filemgr.datatransfer import TransferStatusTracker +from org.apache.oodt.cas.filemgr.structs import Product +from org.apache.oodt.cas.filemgr.util import GenericFileManagerObjectFactory +from org.apache.oodt.cas.filemgr.util import XmlRpcStructFactory as Structs +from org.apache.oodt.cas.metadata import Metadata +from java.lang import Boolean, Double, Integer +from java.util import Hashtable, Vector + +# We choose these default factory classes because it minimizes our dependencies +# on heavyweight external packages, like smelly old SQL databases. +_defaultFactories = { + 'catalog': 'org.apache.oodt.cas.filemgr.catalog.LuceneCatalogFactory', + 'repository': 'org.apache.oodt.cas.filemgr.repository.XMLRepositoryManagerFactory', + 'datatransfer': 'org.apache.oodt.cas.filemgr.datatransfer.LocalDataTransferFactory', + 'validation': 'org.apache.oodt.cas.filemgr.validation.XMLValidationLayerFactory' +} + +# All available permissions. By default, the "root" user will be in the "wheel" +# group and will have these permissions. +_allPerms = [ + 'filemgr.addMetadata', + 'filemgr.addProductReferences', + 'filemgr.addProductType', + 'filemgr.catalogProduct', + 'filemgr.getCurrentFileTransfer', + 'filemgr.getCurrentFileTransfers', + 'filemgr.getElementById', + 'filemgr.getElementByName', + 'filemgr.getElementsByProductType', + 'filemgr.getFirstPage', + 'filemgr.getLastPage', + 'filemgr.getMetadata', + 'filemgr.getNextPage', + 'filemgr.getNumProducts', + 'filemgr.getPrevPage', + 'filemgr.getProductById', + 'filemgr.getProductByName', + 'filemgr.getProductPctTransferred', + 'filemgr.getProductReferences', + 'filemgr.getProductsByProductType', + 'filemgr.getProductTypeById', + 'filemgr.getProductTypeByName', + 'filemgr.getProductTypes', + 'filemgr.getRefPctTransferred', + 'filemgr.getTopNProducts', + 'filemgr.handleRequest', + 'filemgr.hasProduct', + 'filemgr.ingestProduct', + 'filemgr.isTransferComplete', + 'filemgr.pagedQuery', + 'filemgr.query', + 'filemgr.removeFile', + 'filemgr.removeProductTransferStatus', + 'filemgr.setProductTransferStatus', + 'filemgr.transferFile', + 'filemgr.transferringProduct', + 'usermgr.addGroup', + 'usermgr.addPermissionToGroup', + 'usermgr.addUser', + 'usermgr.addUserToGroup', + 'usermgr.removeGroup', + 'usermgr.removePermissionFromGroup', + 'usermgr.removeUser', + 'usermgr.removeUserFromGroup' +] + +def _toJavaBoolean(truthiness): + '''Convert a Python boolean into the string format that Java uses: true or false. + ''' + if truthiness: + return 'true' + else: + return 'false' + + +def _encodePassword(pw): + '''Encode a password using an SHA-1 digest. + ''' + return sha.new(pw).digest() + + +class User: + '''A user of the CAS. Users don't have permissions directly; instead they receive + them implicitly by being members of groups, which do have permissions. + ''' + def __init__(self, userID, name, email, password, groups=[]): + self.userID, self.name, self.email, self.password, self.groups = userID, name, email, password, groups + + def __cmp__(self, other): + return cmp(self.userID, other.userID) + + def __hash__(self): + return hash(self.userID) + + def __repr__(self): + return 'User(userID=%s,name=%s,email=%s,password=%s,groups=%r)' % ( + self.userID, self.name, self.email, self.password, self.groups + ) + + +class Group: + '''A CAS group. The group contains a sequence of permissions, which are strings + that name the XML-RPC methods that the group is allowed to call. + ''' + def __init__(self, groupID, name, email, perms=[]): + self.groupID, self.name, self.email, self.perms = groupID, name, email, perms + + def __cmp__(self, other): + return cmp(self.groupID, other.groupID) + + def __hash__(self): + return hash(self.groupID) + + def __repr__(self): + return 'Group(groupdID=%s,name=%s,email=%s,perms=%r)' % (self.groupdID, self.name, self.email, self.perms) + + +class UserDB: + '''The user database records all the users and groups. + ''' + def __init__(self, users, groups, filename): + self.users, self.groups, self.filename = users, groups, filename + + def authenticate(self, name, password): + '''Authenticate a user by checking the user name and password. + Return true if the user's password matches the given one. The + password given should be in SHA-1 digest format. + ''' + user = self.users[name] + return user.userID == name and user.password == password + + def authorize(self, name, perm): + '''Authorize if the user has the given permission. Return true if + the user can do it, false otherwise. + ''' + user = self.users[name] + for group in user.groups: + if perm in group.perms: + return True + return False + + def save(self): + '''Save the user database to disk. + ''' + f = file(self.filename, 'wb') + pickle.dump(self, f) + f.close() + + +class FileMgrDispatcher(Dispatcher): + '''The file manager dispatcher handles all XML-RPC calls. + ''' + def __init__(self, catalog, repo, xfer, userDB): + self.catalog, self.repo, self.xfer, self.userDB = catalog, repo, xfer, userDB + self.tracker = TransferStatusTracker(self.catalog) + + def handleRequest(self, methodSpecifier, params, user, password): + '''Handle an XML-RPC request. First, authenticate the user. If the user's + authentic (by dint of providing a correct user ID and password pair), then + authorize if the method the user is trying to call is available. + ''' + password = _encodePassword(password) + if self.userDB.authenticate(user, password): + if self.userDB.authorize(user, methodSpecifier): + obj, method = methodSpecifier.split('.') + if obj not in ('filemgr', 'usermgr'): + raise ValueError('Unknown object') + func = getattr(self, method) + return func(params) + raise ValueError('Not authorized for "%s"' % methodSpecifier) + raise ValueError('Illegal user name "%s" and/or password' % user) + + def getProductTypeByName(self, params): + return Result(None, Structs.getXmlRpcProductType(self.repo.getProductTypeByName(params[0]))) + + def ingestProduct(self, params): + productHash, metadata, clientXfer = params + p = Structs.getProductFromXmlRpc(productHash) + p.setTransferStatus(Product.STATUS_TRANSFER) + self.catalog.addProduct(p) + + m = Metadata() + m.addMetadata(metadata) + self.catalog.addMetadata(m, p) + + if not clientXfer: + versioner = GenericFileManagerObjectFactory.getVersionerFromClassName(p.getProductType().getVersioner()) + versioner.createDataStoreReferences(p, m) + self.catalog.addProductReferences(p) + self.xfer.transferProduct(p) + p.setTransferStatus(Product.STATUS_RECEIVED) + self.catalog.setProductTranfserStatus(p) + return Result(None, p.getProductId()) + + def addProductReferences(self, params): + self.catalog.addProductReferences(Structs.getProductFromXmlRpc(params[0])) + return Result(Boolean, 'true') + + def transferringProduct(self, params): + self.tracker.transferringProduct(Structs.getProductFromXmlRpc(params[0])) + return Result(Boolean, 'true') + + def removeProductTransferStatus(self, params): + self.tracker.removeProductTransferStatus(Structs.getProductFromXmlRpc(params[0])) + return Result(Boolean, 'true') + + def setProductTransferStatus(self, params): + self.catalog.setProductTransferStatus(Structs.getProductFromXmlRpc(params[0])) + return Result(Boolean, 'true') + + def getCurrentFileTransfer(self, params): + status = self.tracker.getCurrentFileTransfer() + if status is None: + return Result(None, Hashtable()) + else: + return Result(None, Structs.getXmlRpcFileTransferStatus(status)) + + def getCurrentFileTransfers(self, params): + xfers = self.tracker.getCurrentFileTransfers() + if xfers is not None and len(xfers) > 0: + return Result(None, Structs.getXmlRpcFileTransferStatuses(xfers)) + else: + return Result(None, Vector()) + + def getProductPctTransferred(self, params): + return Result(Double, str(self.tracker.getPctTransferred((Structs.getProductFromXmlRpc(params[0]))))) + + def getRefPctTransferred(self, params): + pct = self.tracker.getPctTransferred(Structs.getReferenceFromXmlRpc(params[0])) + return Result(Double, str(pct)) + + def isTransferComplete(self, params): + return Result(Boolean, _toJavaBoolean(self.tracker.isTransferComplete(Structs.getProductFromXmlRpc(params[0])))) + + def pagedQuery(self, params): + ptype = Structs.getProductTypeFromXmlRpc(params[1]) + query = Structs.getQueryFromXmlRpc(params[0]) + return Result(None, Structs.getXmlRpcProductPage(self.catalog.pagedQuery(query, ptype, params[2]))) + + def getFirstPage(self, params): + ptype = Structs.getProductTypeFromXmlRpc(params[0]) + return Result(None, Structs.getXmlRpcProductPage(self.catalog.getFirstPage(ptype))) + + def getLastPage(self, params): + ptype = Structs.getProductTypeFromXmlRpc(params[0]) + return Result(None, Structs.getXmlRpcProductPage(self.catalog.getLastProductPage(ptype))) + + def getNextPage(self, params): + ptype = Structs.getProductTypeFromXmlRpc(params[0]) + page = Structs.getProductPageFromXmlRpc(params[1]) + return Result(None, Structs.getXmlRpcProductPage(self.catalog.getNextPage(ptype, page))) + + def getPrevPage(self, params): + ptype = Structs.getProductTypeFromXmlRpc(params[0]) + page = Structs.getProductPageFromXmlRpc(params[1]) + return Result(None, Structs.getXmlRpcProductPage(self.catalog.getPrevPage(ptype, page))) + + def addProductType(self, params): + ptype = Structs.getProductTypeFromXmlRpc(params[0]) + self.repo.addProductType(ptype) + return Result(None, ptype.getProductTypeId()) + + def getNumProducts(self, params): + ptype = Structs.getProductTypeFromXmlRpc(params[0]) + return Result(Integer, str(self.catalog.getNumProducts(ptype))) + + def getTopNProducts(self, params): + if len(params) == 1: + return Result(None, Structs.getXmlRpcProductList(self.catalog.getTopNProducts(params[0]))) + ptype = Structs.getProductTypeFromXmlRpc(params[1]) + return Result(None, Structs.getXmlRpcProductList(self.catalog.getTopNProducts(params[0], ptype))) + + def hasProduct(self, params): + p = self.catalog.getProductByName(params[0]) + return Result(Boolean, _toJavaBoolean(p is not None and p.transferStatus == Product.STATUS_RECEIVED)) + + def getMetadata(self, params): + return Result(None, self.catalog.getMetadata(Structs.getProductFromXmlRpc(params[0])).getHashtable()) + + def getProductTypes(self, params): + return Result(None, Structs.getXmlRpcProductList(self.repo.getProductTypes())) + + def getProductReferences(self, params): + p = Structs.getProductFromXmlRpc(params[0]) + return Result(None, Structs.getXmlRpcReferences(self.catalog.getProductReferences(p))) + + def getProductById(self, params): + return Result(None, Structs.getXmlRpcProduct(self.catalog.getProductById(params[0]))) + + def getProductByName(self, params): + return Result(None, Structs.getXmlRpcProduct(self.catalog.getProductByName(params[0]))) + + def getProductsByProductType(self, params): + ptype = Structs.getProductTypeFromXmlRpc(params[0]) + return Result(None, Structs.getXmlRpcProductList(self.catalog.getProductsByProductType(ptype))) + + def getElementsByProductType(self, params): + ptype = Structs.getProductTypeFromXmlRpc(params[0]) + return Structs.getXmlRpcElementList(self.catalog.getValidationLayer().getElements(ptype)) + + def getElementById(self, params): + return Structs.getXmlRpcElement(self.catalog.getValidationLayer().getElementById(params[0])) + + def getElementByName(self, params): + return Structs.getXmlRpcElement(self.catalog.getValidationLayer().getElementByName(params[0])) + + def query(self, params): + q = Structs.getQueryFromXmlRpc(params[0]) + ptype = Structs.getProductTypeFromXmlRpc(params[1]) + ids = self.catalog.query(q, ptype) + if ids is not None and len(ids) > 0: + return Result(None, [self.catalog.getProductById(i) for i in ids]) + return Result(None, Vector()) + + def getProductTypeById(self, params): + ptype = self.repo.getProductTypeById(params[0]) + return Result(None, Structs.getXmlRpcProductType(ptype)) + + def catalogProduct(self, params): + p = Structs.getProductFromXmlRpc(params[0]) + return Result(None, self.catalog.addProduct(p)) + + def addMetadata(self, params): + p = Structs.getProductFromXmlRpc(params[0]) + m = Metadata() + m.addMetadata(params[1]) + self.catalog.addMetadata(m, p) + return Result(Boolean, 'true') + + def transferFile(self, params): + outFile, data, offset, numBytes = params + if os.path.exists(outFile): + out = file(outFile, 'ab') + else: + dirPath = os.dirname(outFile) + os.makedirs(dirPath) + out = file(outFile, 'wb') + out.seek(offset) + out.write(data) + out.close() + return Result(Boolean, 'true') + + def removeFile(self, params): + os.remove(params[0]) + return Result(Boolean, 'true') + + def addUser(self, params): + userID = params[0] + user = User(userID, params[1], params[2], _encodePassword(params[3]), []) + self.userDB.users[userID] = user + self.userDB.save() + return Result(Boolean, 'true') + + def removeUser(self, params): + del self.userDB.users[params[0]] + return Result(Boolean, 'true') + + def addGroup(self, params): + groupID = params[0] + group = Group(groupID, params[1], params[2]) + self.userDB.groups[groupID] = group + self.userDB.save() + return Result(Boolean, 'true') + + def removeGroup(self, params): + groupID = params[0] + del self.userDB.groups[groupID] + for user in self.userDB.users.itervalues(): + indexes = [] + index = 0 + for group in user.groups: + if group.groupID == groupID: + indexes.append(index) + index += 1 + indexes.reverse() + for index in indexes: + del user.groups[index] + self.userDB.save() + return Result(Boolean, 'true') + + def addUserToGroup(self, params): + self.userDB.users[params[0]].groups.append(self.userDB.groups[params[1]]) + return Result(Boolean, 'true') + + def removeUserFromGroup(self, params): + groupID = params[1] + user = self.userDB.users[params[0]] + indexes = [] + index = 0 + for group in user.group: + if group.groupID == groupID: + indexes.append(index) + index += 1 + indexes.reverse() + for index in indexes: + del user.groups[index] + self.userDB.save() + return Result(Boolean, 'true') + + def addPermissionToGroup(self, params): + self.userDB.groups[params[0]].perms.append(params[1]) + self.userDB.save() + return Result(Boolean, 'true') + + def removePermissionFromGroup(self, params): + permName = params[1] + group = self.userDB.groups[params[0]] + indexes = [] + index = 0 + for perm in group.perms: + if perm == permName: + indexes.append(index) + indexes.reverse() + for index in indexes: + del group.perms[index] + self.userDB.save() + return Result(Boolean, 'true') + + +def _usage(): + '''Show a usage message to the stderr and quit. + ''' + print >>sys.stderr, 'Usage: %s [-c <configFile>]' % sys.argv[0] + print >>sys.stderr, ' or: %s [--config=<configFile>]' % sys.argv[0] + sys.exit(2) + + +def _parseCommandLine(): + '''Parse the command line options. If any. The only option is -c (or --config) + that names a configuration file to use. If none given, reasonable defaults are + used. Well, mostly reasonable. + ''' + try: + opts, args = getopt.getopt(sys.argv[1:], 'c:', 'config=') + except getopt.GetoptError: + _usage() + configFile = None + for option, arg in opts: + if option in ('-c', '--config'): + configFile = arg + if configFile is None or len(configFile) == 0: + _usage() + return configFile + + +def _getConfig(configFile): + '''Get the configuration. This populates a configuration with default values + and then overwrites them with the configFile, which may be None (in which case, + no overwriting happens). + ''' + configParser = ConfigParser() + configParser.add_section('factories') + for key, val in _defaultFactories.iteritems(): + configParser.set('factories', key, val) + + configParser.add_section('index') + configParser.set('index', 'path', 'index') + configParser.set('index', 'pageSize', '20') + + current = '/'.join(os.path.split(os.getcwd())) + configParser.add_section('policies') + configParser.set('policies', 'repo', 'file:%s/policy' % current) + configParser.set('policies', 'validation', 'file:%s/policy' % current) + configParser.set('policies', 'user', '%s/user.db' % current) + + if configFile is not None: + configParser.readfp(file(configFile)) + return configParser + + +def _setJavaProperties(config): + '''Set Java-based properties. The Java-based cas expects a whole bunch of + system properties to be set, sort of like global variables. Woot! Global + variables! + ''' + from java.lang import System + System.setProperty('org.apache.oodt.cas.filemgr.catalog.lucene.idxPath', config.get('index', 'path')) + System.setProperty('org.apache.oodt.cas.filemgr.catalog.lucene.pageSize', config.get('index', 'pageSize')) + System.setProperty('org.apache.oodt.cas.filemgr.repositorymgr.dirs', config.get('policies', 'repo')) + System.setProperty('org.apache.oodt.cas.filemgr.validation.dirs', config.get('policies', 'validation')) + System.setProperty('org.apache.oodt.cas.filemgr.datatransfer.remote.chunkSize', '1024') + System.setProperty('filemgr.repository.factory', config.get('factories', 'repository')) + System.setProperty('filemgr.catalog.factory', config.get('factories', 'catalog')) + System.setProperty('filemgr.datatransfer.factory', config.get('factories', 'datatransfer')) + System.setProperty('filemgr.validationLayer.factory', config.get('factories', 'validation')) + +def _getUserDB(path): + '''Get the user database, creating it if necessary. + ''' + try: + f = file(path, 'rb') + db = pickle.load(f) + f.close() + db.filename = path + db.save() + except: + wheel = Group('wheel', 'Administrators', '[email protected]', _allPerms) + root = User('root', 'Super User', '[email protected]', _encodePassword('poipu'), [wheel]) + db = UserDB({'root': root}, {'wheel': wheel}, path) + db.save() + return db + + +def main(): + '''Start the CAS Filemgr Backend. + ''' + configFile = _parseCommandLine() + configParser = _getConfig(configFile) + _setJavaProperties(configParser) + + catalog = GenericFileManagerObjectFactory.getCatalogServiceFromFactory(configParser.get('factories', 'catalog')) + repo = GenericFileManagerObjectFactory.getRepositoryManagerServiceFromFactory(configParser.get('factories', 'repository')) + xfer = GenericFileManagerObjectFactory.getDataTransferServiceFromFactory(configParser.get('factories', 'datatransfer')) + userDB = _getUserDB(configParser.get('policies', 'user')) + + ws = SecureWebServer(1999) + ws.addDispatcher(FileMgrDispatcher(catalog, repo, xfer, userDB)) + ws.start() + + +if __name__ == '__main__': + main() + http://git-wip-us.apache.org/repos/asf/oodt/blob/d329e1ab/agility/oodt/metadata.py ---------------------------------------------------------------------- diff --git a/agility/oodt/metadata.py b/agility/oodt/metadata.py new file mode 100644 index 0000000..1f1a9a9 --- /dev/null +++ b/agility/oodt/metadata.py @@ -0,0 +1,29 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF 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. + + +class MetaData(object): + def __init__(self): + self.data = {} + + def addMetaData(self, key, value): + if key not in self.data: + self.data[key] = [] + self.data[key].append(value) + + def toXmlRpc(self): + return self.data http://git-wip-us.apache.org/repos/asf/oodt/blob/d329e1ab/agility/oodt/workflow.py ---------------------------------------------------------------------- diff --git a/agility/oodt/workflow.py b/agility/oodt/workflow.py new file mode 100644 index 0000000..ae83c23 --- /dev/null +++ b/agility/oodt/workflow.py @@ -0,0 +1,134 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF 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. + + +import xmlrpclib +from oodt_metadata import MetaData + + +class OODTWorkFlowWrapper(object): + def __init__(self, host): + self._server = xmlrpclib.Server(host) + + #String executeDynamicWorkflow(Vector<String> taskIds, Hashtable metadata) + def executeDynamicWorkflow(self, taskIds, metadata): + return self._server.workflowmgr.executeDynamicWorkflow(taskIds, metadata) + #Hashtable getConditionById(String conditionId) + def getConditionById(self, conditionId): + return self._server.workflowmgr.getConditionById(conditionId) + #Hashtable getFirstPage() + def getFirstPage(self): + return self._server.workflowmgr.getFirstPage() + #Hashtable getLastPage() + def getLastPage(self): + return self._server.workflowmgr.getLastPage() + #Hashtable getNextPage(Hashtable currentPage) + def getNextPage(self, currentPage): + return self._server.workflowmgr.getNextPage() + #int getNumWorkflowInstances() + def getNumWorkflowInstances(self): + return self._server.workflowmgr.getNumWorkflowInstances() + #int getNumWorkflowInstancesByStatus(String status) + def getNumWorkflowInstancesByStatus(self, status): + return self._server.workflowmgr.getNumWorkflowInstancesByStatus(status) + #Hashtable getPrevPage(Hashtable currentPage) + def getPrevPage(self, currentPage): + return self._server.workflowmgr.getPrevPage(currentPage) + #Vector getRegisteredEvents() + def getEventNames(self): + return self._server.workflowmgr.getRegisteredEvents() + #Hashtable getTaskById(String taskId) + def getTaskById(self, taskId): + return self._server.workflowmgr.getTaskById(taskId) + #Hashtable getWorkflowById(String workflowId) + def getWorkflowById(self, workflowId): + return self._server.workflowmgr.getWorkflowById(workflowId) + #double getWorkflowCurrentTaskWallClockMinutes(String workflowInstId) + def getWorkflowCurrentTaskWallClockMinutes(self, workflowInstId): + return self._server.workflowmgr.getWorkflowCurrentTaskWallClockMinutes(workflowInstId) + #Hashtable getWorkflowInstanceById(String wInstId) + def getWorkflowInstanceById(self,wInstId): + return self._server.workflowmgr.getWorkflowInstanceById(wInstId) + #Hashtable getWorkflowInstanceMetadata(String wInstId) + def getWorkflowInstanceMetadata(self, wInstId): + return self._server.workflowmgr.getWorkflowInstanceMetadata(wInstId) + #Vector getWorkflowInstances() + def getWorkflowInstances(self): + return self._server.workflowmgr.getWorkflowInstances() + #Vector getWorkflowInstancesByStatus(String status) + def getWorkflowInstancesByStatus(self, status): + return self._server.workflowmgr.getWorkflowInstancesByStatus(status) + #Vector getWorkflows() + def getWorkflows(self): + return self._server.workflowmgr.getWorkflows() + #Vector getWorkflowsByEvent(String eventName) + def getWorkflowsByEvent(self, eventName): + return self._server.workflowmgr.getWorkflowsByEvent(eventName) + #double getWorkflowWallClockMinutes(String workflowInstId) + def getWorkflowWallClockMinutes(self,workflowInstId): + return getWorkflowWallClockMinutes(workflowInstId) + #boolean handleEvent(String eventName, Hashtable metadata) + def startEvent(self, eventName, metaData): + return self._server.workflowmgr.handleEvent(eventName, metaData) + #Hashtable paginateWorkflowInstances(int pageNum) + def paginateWorkflowInstances(self, pageNum): + return self._server.workflowmgr.paginateWorkflowInstances(pageNum) + #Hashtable paginateWorkflowInstances(int pageNum, String status)(self, workflowInstId) + def paginateWorkflowInstances(self, pageNum, status): + return self._server.workflowmgr.paginateWorkflowInstances(pageNum, status) + #boolean pauseWorkflowInstance(String workflowInstId) + def pauseWorkflowInstance(self, workflowInstId): + return self._server.workflowmgr.pauseWorkflowInstance(workflowInstId) + #boolean resumeWorkflowInstance(String workflowInstId) + def resumeWorkflowInstance(self, workflowInstId): + return self._server.workflowmgr.resumeWorkflowInstance(workflowInstId) + #boolean setWorkflowInstanceCurrentTaskEndDateTime(String wInstId, String endDateTimeIsoStr) + def setWorkflowInstanceCurrentTaskEndDateTime(self, workflowInstId, endDateTimeIsoStr): + return self._server.workflowmgr.setWorkflowInstanceCurrentTaskEndDateTime(workflowInstId, endDateTimeIsoStr) + #boolean setWorkflowInstanceCurrentTaskStartDateTime(String wInstId, String startDateTimeIsoStr) + def setWorkflowInstanceCurrentTaskStartDateTime(self, workflowInstId, startDateTimeIsoStr): + return self._server.workflowmgr.setWorkflowInstanceCurrentTaskStartDateTime(workflowInstId, startDateTimeIsoStr) + #boolean stopWorkflowInstance(String workflowInstId) + def stopWorkflowInstance(self, workflowInstId): + return self._server.workflowmgr.stopWorkflowInstance(workflowInstId) + #boolean updateMetadataForWorkflow(String workflowInstId, Hashtable metadata) + def updateMetadataForWorkflow(self, workflowInstId, metadata): + return self._server.workflowmgr.updateMetadataForWorkflow(workflowInstId, metadata) + #boolean updateWorkflowInstance(Hashtable workflowInst) + def updateWorkflowInstance(workflowInst): + return self._server.workflowmgr.updateWorkflowInstance(workflowInst) + #boolean updateWorkflowInstanceStatus(String workflowInstanceId, String status) + def updateWorkflowInstanceStatus(self, workflowInstId, status): + return self._server.workflowmgr.updateWorkflowInstanceStatus(workflowInstId, status) + + + +def main(): + # create instance + oodt = OODTWorkFlowWrapper("http://localhost:9200";) + # get event info + events = oodt.getEventNames() + # create metadata object to invoke an event + met = MetaData() + met.addMetaData("hello", "world") + # print available events + print 'available events:', events + + # oodt.startEvent(events[0], met.toXmlRpc()) + +if __name__ == '__main__': + main() http://git-wip-us.apache.org/repos/asf/oodt/blob/d329e1ab/filemgr/src/main/python/fm.conf ---------------------------------------------------------------------- diff --git a/filemgr/src/main/python/fm.conf b/filemgr/src/main/python/fm.conf deleted file mode 100644 index dae6438..0000000 --- a/filemgr/src/main/python/fm.conf +++ /dev/null @@ -1,43 +0,0 @@ -# encoding: utf-8 -#/* -# * Licensed to the Apache Software Foundation (ASF) under one or more -# * contributor license agreements. See the NOTICE file distributed with -# * this work for additional information regarding copyright ownership. -# * The ASF 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. -# */ - -[index] -# The path gives the name of the directory where the Lucene search engine -# will build its indexes. The path refers to a directory; it will be created -# if it doesn't exist. If it does exist, it must refer to an -# existing Lucene index with correct sub-files present. So, don't make this -# directory manually, or you'll be confused by error messages! -path = @INDEX_PATH@ -pageSize = 20 - -[policies] -# The 'repo' and 'validation' are comma-separated lists of URLs. Typically, they -# contain a single "file:" style URL that names a directory with policy XML files -# within. The 'user' identifies the user database. It'll be created if it -# doesn't exist. -repo = @REPO_POLICY_URLS@ -validation = @VALIDATION_POLICY_URLS@ -user = @USER_DB_PATH@ - -[factories] -# There's no need to change anything here unless you really know what you're doing, -# and trust me, you don't. Don't take it personally. It's just a fact. -catalog = org.apache.oodt.cas.filemgr.catalog.LuceneCatalogFactory -repository = org.apache.oodt.cas.filemgr.repository.XMLRepositoryManagerFactory -datatransfer = org.apache.oodt.cas.filemgr.datatransfer.LocalDataTransferFactory -validation =org.apache.oodt.cas.filemgr.validation.XMLValidationLayerFactory http://git-wip-us.apache.org/repos/asf/oodt/blob/d329e1ab/filemgr/src/main/python/fm.py ---------------------------------------------------------------------- diff --git a/filemgr/src/main/python/fm.py b/filemgr/src/main/python/fm.py deleted file mode 100644 index 5eed046..0000000 --- a/filemgr/src/main/python/fm.py +++ /dev/null @@ -1,554 +0,0 @@ -#!/usr/bin/env python -# encoding: utf-8 -#/* -# * Licensed to the Apache Software Foundation (ASF) under one or more -# * contributor license agreements. See the NOTICE file distributed with -# * this work for additional information regarding copyright ownership. -# * The ASF 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. -# */ -''' -CAS Filemgr Python Server - -This is the Python server API for an authenticated Catalog and Archive System. It uses the -OODT Catalog and Archive System as its core, adding user/password-based authentication -and role-based authorization. -''' - -import sys, os, os.path -import getopt, sha, pickle - -from ConfigParser import ConfigParser -from org.apache.oodt.cas.filemgr.system.auth import SecureWebServer, Dispatcher, Result -from org.apache.oodt.cas.filemgr.datatransfer import TransferStatusTracker -from org.apache.oodt.cas.filemgr.structs import Product -from org.apache.oodt.cas.filemgr.util import GenericFileManagerObjectFactory -from org.apache.oodt.cas.filemgr.util import XmlRpcStructFactory as Structs -from org.apache.oodt.cas.metadata import Metadata -from java.lang import Boolean, Double, Integer -from java.util import Hashtable, Vector - -# We choose these default factory classes because it minimizes our dependencies -# on heavyweight external packages, like smelly old SQL databases. -_defaultFactories = { - 'catalog': 'org.apache.oodt.cas.filemgr.catalog.LuceneCatalogFactory', - 'repository': 'org.apache.oodt.cas.filemgr.repository.XMLRepositoryManagerFactory', - 'datatransfer': 'org.apache.oodt.cas.filemgr.datatransfer.LocalDataTransferFactory', - 'validation': 'org.apache.oodt.cas.filemgr.validation.XMLValidationLayerFactory' -} - -# All available permissions. By default, the "root" user will be in the "wheel" -# group and will have these permissions. -_allPerms = [ - 'filemgr.addMetadata', - 'filemgr.addProductReferences', - 'filemgr.addProductType', - 'filemgr.catalogProduct', - 'filemgr.getCurrentFileTransfer', - 'filemgr.getCurrentFileTransfers', - 'filemgr.getElementById', - 'filemgr.getElementByName', - 'filemgr.getElementsByProductType', - 'filemgr.getFirstPage', - 'filemgr.getLastPage', - 'filemgr.getMetadata', - 'filemgr.getNextPage', - 'filemgr.getNumProducts', - 'filemgr.getPrevPage', - 'filemgr.getProductById', - 'filemgr.getProductByName', - 'filemgr.getProductPctTransferred', - 'filemgr.getProductReferences', - 'filemgr.getProductsByProductType', - 'filemgr.getProductTypeById', - 'filemgr.getProductTypeByName', - 'filemgr.getProductTypes', - 'filemgr.getRefPctTransferred', - 'filemgr.getTopNProducts', - 'filemgr.handleRequest', - 'filemgr.hasProduct', - 'filemgr.ingestProduct', - 'filemgr.isTransferComplete', - 'filemgr.pagedQuery', - 'filemgr.query', - 'filemgr.removeFile', - 'filemgr.removeProductTransferStatus', - 'filemgr.setProductTransferStatus', - 'filemgr.transferFile', - 'filemgr.transferringProduct', - 'usermgr.addGroup', - 'usermgr.addPermissionToGroup', - 'usermgr.addUser', - 'usermgr.addUserToGroup', - 'usermgr.removeGroup', - 'usermgr.removePermissionFromGroup', - 'usermgr.removeUser', - 'usermgr.removeUserFromGroup' -] - -def _toJavaBoolean(truthiness): - '''Convert a Python boolean into the string format that Java uses: true or false. - ''' - if truthiness: - return 'true' - else: - return 'false' - - -def _encodePassword(pw): - '''Encode a password using an SHA-1 digest. - ''' - return sha.new(pw).digest() - - -class User: - '''A user of the CAS. Users don't have permissions directly; instead they receive - them implicitly by being members of groups, which do have permissions. - ''' - def __init__(self, userID, name, email, password, groups=[]): - self.userID, self.name, self.email, self.password, self.groups = userID, name, email, password, groups - - def __cmp__(self, other): - return cmp(self.userID, other.userID) - - def __hash__(self): - return hash(self.userID) - - def __repr__(self): - return 'User(userID=%s,name=%s,email=%s,password=%s,groups=%r)' % ( - self.userID, self.name, self.email, self.password, self.groups - ) - - -class Group: - '''A CAS group. The group contains a sequence of permissions, which are strings - that name the XML-RPC methods that the group is allowed to call. - ''' - def __init__(self, groupID, name, email, perms=[]): - self.groupID, self.name, self.email, self.perms = groupID, name, email, perms - - def __cmp__(self, other): - return cmp(self.groupID, other.groupID) - - def __hash__(self): - return hash(self.groupID) - - def __repr__(self): - return 'Group(groupdID=%s,name=%s,email=%s,perms=%r)' % (self.groupdID, self.name, self.email, self.perms) - - -class UserDB: - '''The user database records all the users and groups. - ''' - def __init__(self, users, groups, filename): - self.users, self.groups, self.filename = users, groups, filename - - def authenticate(self, name, password): - '''Authenticate a user by checking the user name and password. - Return true if the user's password matches the given one. The - password given should be in SHA-1 digest format. - ''' - user = self.users[name] - return user.userID == name and user.password == password - - def authorize(self, name, perm): - '''Authorize if the user has the given permission. Return true if - the user can do it, false otherwise. - ''' - user = self.users[name] - for group in user.groups: - if perm in group.perms: - return True - return False - - def save(self): - '''Save the user database to disk. - ''' - f = file(self.filename, 'wb') - pickle.dump(self, f) - f.close() - - -class FileMgrDispatcher(Dispatcher): - '''The file manager dispatcher handles all XML-RPC calls. - ''' - def __init__(self, catalog, repo, xfer, userDB): - self.catalog, self.repo, self.xfer, self.userDB = catalog, repo, xfer, userDB - self.tracker = TransferStatusTracker(self.catalog) - - def handleRequest(self, methodSpecifier, params, user, password): - '''Handle an XML-RPC request. First, authenticate the user. If the user's - authentic (by dint of providing a correct user ID and password pair), then - authorize if the method the user is trying to call is available. - ''' - password = _encodePassword(password) - if self.userDB.authenticate(user, password): - if self.userDB.authorize(user, methodSpecifier): - obj, method = methodSpecifier.split('.') - if obj not in ('filemgr', 'usermgr'): - raise ValueError('Unknown object') - func = getattr(self, method) - return func(params) - raise ValueError('Not authorized for "%s"' % methodSpecifier) - raise ValueError('Illegal user name "%s" and/or password' % user) - - def getProductTypeByName(self, params): - return Result(None, Structs.getXmlRpcProductType(self.repo.getProductTypeByName(params[0]))) - - def ingestProduct(self, params): - productHash, metadata, clientXfer = params - p = Structs.getProductFromXmlRpc(productHash) - p.setTransferStatus(Product.STATUS_TRANSFER) - self.catalog.addProduct(p) - - m = Metadata() - m.addMetadata(metadata) - self.catalog.addMetadata(m, p) - - if not clientXfer: - versioner = GenericFileManagerObjectFactory.getVersionerFromClassName(p.getProductType().getVersioner()) - versioner.createDataStoreReferences(p, m) - self.catalog.addProductReferences(p) - self.xfer.transferProduct(p) - p.setTransferStatus(Product.STATUS_RECEIVED) - self.catalog.setProductTranfserStatus(p) - return Result(None, p.getProductId()) - - def addProductReferences(self, params): - self.catalog.addProductReferences(Structs.getProductFromXmlRpc(params[0])) - return Result(Boolean, 'true') - - def transferringProduct(self, params): - self.tracker.transferringProduct(Structs.getProductFromXmlRpc(params[0])) - return Result(Boolean, 'true') - - def removeProductTransferStatus(self, params): - self.tracker.removeProductTransferStatus(Structs.getProductFromXmlRpc(params[0])) - return Result(Boolean, 'true') - - def setProductTransferStatus(self, params): - self.catalog.setProductTransferStatus(Structs.getProductFromXmlRpc(params[0])) - return Result(Boolean, 'true') - - def getCurrentFileTransfer(self, params): - status = self.tracker.getCurrentFileTransfer() - if status is None: - return Result(None, Hashtable()) - else: - return Result(None, Structs.getXmlRpcFileTransferStatus(status)) - - def getCurrentFileTransfers(self, params): - xfers = self.tracker.getCurrentFileTransfers() - if xfers is not None and len(xfers) > 0: - return Result(None, Structs.getXmlRpcFileTransferStatuses(xfers)) - else: - return Result(None, Vector()) - - def getProductPctTransferred(self, params): - return Result(Double, str(self.tracker.getPctTransferred((Structs.getProductFromXmlRpc(params[0]))))) - - def getRefPctTransferred(self, params): - pct = self.tracker.getPctTransferred(Structs.getReferenceFromXmlRpc(params[0])) - return Result(Double, str(pct)) - - def isTransferComplete(self, params): - return Result(Boolean, _toJavaBoolean(self.tracker.isTransferComplete(Structs.getProductFromXmlRpc(params[0])))) - - def pagedQuery(self, params): - ptype = Structs.getProductTypeFromXmlRpc(params[1]) - query = Structs.getQueryFromXmlRpc(params[0]) - return Result(None, Structs.getXmlRpcProductPage(self.catalog.pagedQuery(query, ptype, params[2]))) - - def getFirstPage(self, params): - ptype = Structs.getProductTypeFromXmlRpc(params[0]) - return Result(None, Structs.getXmlRpcProductPage(self.catalog.getFirstPage(ptype))) - - def getLastPage(self, params): - ptype = Structs.getProductTypeFromXmlRpc(params[0]) - return Result(None, Structs.getXmlRpcProductPage(self.catalog.getLastProductPage(ptype))) - - def getNextPage(self, params): - ptype = Structs.getProductTypeFromXmlRpc(params[0]) - page = Structs.getProductPageFromXmlRpc(params[1]) - return Result(None, Structs.getXmlRpcProductPage(self.catalog.getNextPage(ptype, page))) - - def getPrevPage(self, params): - ptype = Structs.getProductTypeFromXmlRpc(params[0]) - page = Structs.getProductPageFromXmlRpc(params[1]) - return Result(None, Structs.getXmlRpcProductPage(self.catalog.getPrevPage(ptype, page))) - - def addProductType(self, params): - ptype = Structs.getProductTypeFromXmlRpc(params[0]) - self.repo.addProductType(ptype) - return Result(None, ptype.getProductTypeId()) - - def getNumProducts(self, params): - ptype = Structs.getProductTypeFromXmlRpc(params[0]) - return Result(Integer, str(self.catalog.getNumProducts(ptype))) - - def getTopNProducts(self, params): - if len(params) == 1: - return Result(None, Structs.getXmlRpcProductList(self.catalog.getTopNProducts(params[0]))) - ptype = Structs.getProductTypeFromXmlRpc(params[1]) - return Result(None, Structs.getXmlRpcProductList(self.catalog.getTopNProducts(params[0], ptype))) - - def hasProduct(self, params): - p = self.catalog.getProductByName(params[0]) - return Result(Boolean, _toJavaBoolean(p is not None and p.transferStatus == Product.STATUS_RECEIVED)) - - def getMetadata(self, params): - return Result(None, self.catalog.getMetadata(Structs.getProductFromXmlRpc(params[0])).getHashtable()) - - def getProductTypes(self, params): - return Result(None, Structs.getXmlRpcProductList(self.repo.getProductTypes())) - - def getProductReferences(self, params): - p = Structs.getProductFromXmlRpc(params[0]) - return Result(None, Structs.getXmlRpcReferences(self.catalog.getProductReferences(p))) - - def getProductById(self, params): - return Result(None, Structs.getXmlRpcProduct(self.catalog.getProductById(params[0]))) - - def getProductByName(self, params): - return Result(None, Structs.getXmlRpcProduct(self.catalog.getProductByName(params[0]))) - - def getProductsByProductType(self, params): - ptype = Structs.getProductTypeFromXmlRpc(params[0]) - return Result(None, Structs.getXmlRpcProductList(self.catalog.getProductsByProductType(ptype))) - - def getElementsByProductType(self, params): - ptype = Structs.getProductTypeFromXmlRpc(params[0]) - return Structs.getXmlRpcElementList(self.catalog.getValidationLayer().getElements(ptype)) - - def getElementById(self, params): - return Structs.getXmlRpcElement(self.catalog.getValidationLayer().getElementById(params[0])) - - def getElementByName(self, params): - return Structs.getXmlRpcElement(self.catalog.getValidationLayer().getElementByName(params[0])) - - def query(self, params): - q = Structs.getQueryFromXmlRpc(params[0]) - ptype = Structs.getProductTypeFromXmlRpc(params[1]) - ids = self.catalog.query(q, ptype) - if ids is not None and len(ids) > 0: - return Result(None, [self.catalog.getProductById(i) for i in ids]) - return Result(None, Vector()) - - def getProductTypeById(self, params): - ptype = self.repo.getProductTypeById(params[0]) - return Result(None, Structs.getXmlRpcProductType(ptype)) - - def catalogProduct(self, params): - p = Structs.getProductFromXmlRpc(params[0]) - return Result(None, self.catalog.addProduct(p)) - - def addMetadata(self, params): - p = Structs.getProductFromXmlRpc(params[0]) - m = Metadata() - m.addMetadata(params[1]) - self.catalog.addMetadata(m, p) - return Result(Boolean, 'true') - - def transferFile(self, params): - outFile, data, offset, numBytes = params - if os.path.exists(outFile): - out = file(outFile, 'ab') - else: - dirPath = os.dirname(outFile) - os.makedirs(dirPath) - out = file(outFile, 'wb') - out.seek(offset) - out.write(data) - out.close() - return Result(Boolean, 'true') - - def removeFile(self, params): - os.remove(params[0]) - return Result(Boolean, 'true') - - def addUser(self, params): - userID = params[0] - user = User(userID, params[1], params[2], _encodePassword(params[3]), []) - self.userDB.users[userID] = user - self.userDB.save() - return Result(Boolean, 'true') - - def removeUser(self, params): - del self.userDB.users[params[0]] - return Result(Boolean, 'true') - - def addGroup(self, params): - groupID = params[0] - group = Group(groupID, params[1], params[2]) - self.userDB.groups[groupID] = group - self.userDB.save() - return Result(Boolean, 'true') - - def removeGroup(self, params): - groupID = params[0] - del self.userDB.groups[groupID] - for user in self.userDB.users.itervalues(): - indexes = [] - index = 0 - for group in user.groups: - if group.groupID == groupID: - indexes.append(index) - index += 1 - indexes.reverse() - for index in indexes: - del user.groups[index] - self.userDB.save() - return Result(Boolean, 'true') - - def addUserToGroup(self, params): - self.userDB.users[params[0]].groups.append(self.userDB.groups[params[1]]) - return Result(Boolean, 'true') - - def removeUserFromGroup(self, params): - groupID = params[1] - user = self.userDB.users[params[0]] - indexes = [] - index = 0 - for group in user.group: - if group.groupID == groupID: - indexes.append(index) - index += 1 - indexes.reverse() - for index in indexes: - del user.groups[index] - self.userDB.save() - return Result(Boolean, 'true') - - def addPermissionToGroup(self, params): - self.userDB.groups[params[0]].perms.append(params[1]) - self.userDB.save() - return Result(Boolean, 'true') - - def removePermissionFromGroup(self, params): - permName = params[1] - group = self.userDB.groups[params[0]] - indexes = [] - index = 0 - for perm in group.perms: - if perm == permName: - indexes.append(index) - indexes.reverse() - for index in indexes: - del group.perms[index] - self.userDB.save() - return Result(Boolean, 'true') - - -def _usage(): - '''Show a usage message to the stderr and quit. - ''' - print >>sys.stderr, 'Usage: %s [-c <configFile>]' % sys.argv[0] - print >>sys.stderr, ' or: %s [--config=<configFile>]' % sys.argv[0] - sys.exit(2) - - -def _parseCommandLine(): - '''Parse the command line options. If any. The only option is -c (or --config) - that names a configuration file to use. If none given, reasonable defaults are - used. Well, mostly reasonable. - ''' - try: - opts, args = getopt.getopt(sys.argv[1:], 'c:', 'config=') - except getopt.GetoptError: - _usage() - configFile = None - for option, arg in opts: - if option in ('-c', '--config'): - configFile = arg - if configFile is None or len(configFile) == 0: - _usage() - return configFile - - -def _getConfig(configFile): - '''Get the configuration. This populates a configuration with default values - and then overwrites them with the configFile, which may be None (in which case, - no overwriting happens). - ''' - configParser = ConfigParser() - configParser.add_section('factories') - for key, val in _defaultFactories.iteritems(): - configParser.set('factories', key, val) - - configParser.add_section('index') - configParser.set('index', 'path', 'index') - configParser.set('index', 'pageSize', '20') - - current = '/'.join(os.path.split(os.getcwd())) - configParser.add_section('policies') - configParser.set('policies', 'repo', 'file:%s/policy' % current) - configParser.set('policies', 'validation', 'file:%s/policy' % current) - configParser.set('policies', 'user', '%s/user.db' % current) - - if configFile is not None: - configParser.readfp(file(configFile)) - return configParser - - -def _setJavaProperties(config): - '''Set Java-based properties. The Java-based cas expects a whole bunch of - system properties to be set, sort of like global variables. Woot! Global - variables! - ''' - from java.lang import System - System.setProperty('org.apache.oodt.cas.filemgr.catalog.lucene.idxPath', config.get('index', 'path')) - System.setProperty('org.apache.oodt.cas.filemgr.catalog.lucene.pageSize', config.get('index', 'pageSize')) - System.setProperty('org.apache.oodt.cas.filemgr.repositorymgr.dirs', config.get('policies', 'repo')) - System.setProperty('org.apache.oodt.cas.filemgr.validation.dirs', config.get('policies', 'validation')) - System.setProperty('org.apache.oodt.cas.filemgr.datatransfer.remote.chunkSize', '1024') - System.setProperty('filemgr.repository.factory', config.get('factories', 'repository')) - System.setProperty('filemgr.catalog.factory', config.get('factories', 'catalog')) - System.setProperty('filemgr.datatransfer.factory', config.get('factories', 'datatransfer')) - System.setProperty('filemgr.validationLayer.factory', config.get('factories', 'validation')) - -def _getUserDB(path): - '''Get the user database, creating it if necessary. - ''' - try: - f = file(path, 'rb') - db = pickle.load(f) - f.close() - db.filename = path - db.save() - except: - wheel = Group('wheel', 'Administrators', '[email protected]', _allPerms) - root = User('root', 'Super User', '[email protected]', _encodePassword('poipu'), [wheel]) - db = UserDB({'root': root}, {'wheel': wheel}, path) - db.save() - return db - - -def main(): - '''Start the CAS Filemgr Backend. - ''' - configFile = _parseCommandLine() - configParser = _getConfig(configFile) - _setJavaProperties(configParser) - - catalog = GenericFileManagerObjectFactory.getCatalogServiceFromFactory(configParser.get('factories', 'catalog')) - repo = GenericFileManagerObjectFactory.getRepositoryManagerServiceFromFactory(configParser.get('factories', 'repository')) - xfer = GenericFileManagerObjectFactory.getDataTransferServiceFromFactory(configParser.get('factories', 'datatransfer')) - userDB = _getUserDB(configParser.get('policies', 'user')) - - ws = SecureWebServer(1999) - ws.addDispatcher(FileMgrDispatcher(catalog, repo, xfer, userDB)) - ws.start() - - -if __name__ == '__main__': - main() - http://git-wip-us.apache.org/repos/asf/oodt/blob/d329e1ab/filemgr/src/main/python/setclasspath.sh ---------------------------------------------------------------------- diff --git a/filemgr/src/main/python/setclasspath.sh b/filemgr/src/main/python/setclasspath.sh deleted file mode 100755 index 1fbfc48..0000000 --- a/filemgr/src/main/python/setclasspath.sh +++ /dev/null @@ -1,17 +0,0 @@ -#/* -# * Licensed to the Apache Software Foundation (ASF) under one or more -# * contributor license agreements. See the NOTICE file distributed with -# * this work for additional information regarding copyright ownership. -# * The ASF 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. -# */ -setenv CLASSPATH ${HOME}/.m2/repository/org.apache.oodt.cas-filemgr/1.7.0-dev/cas-filemgr-1.7.0-dev.jar:${HOME}/.m2/repository/xmlrpc/xmlrpc/2.1-dev/xmlrpc-2.1-dev.jar:${HOME}/.m2/repository/org.apache.oodt.cas-metadata/1.5.0/cas-metadata-1.5.0.jar:${HOME}/.m2/repository/org/apache/lucene/lucene-core/2.0.0/lucene-core-2.0.0.jar:${HOME}/.m2/repository/jug/jug-asl/2.0.0/jug-asl-2.0.0.jar http://git-wip-us.apache.org/repos/asf/oodt/blob/d329e1ab/workflow/src/main/python/xmlrpc-wrapper/oodt_metadata.py ---------------------------------------------------------------------- diff --git a/workflow/src/main/python/xmlrpc-wrapper/oodt_metadata.py b/workflow/src/main/python/xmlrpc-wrapper/oodt_metadata.py deleted file mode 100644 index 1f1a9a9..0000000 --- a/workflow/src/main/python/xmlrpc-wrapper/oodt_metadata.py +++ /dev/null @@ -1,29 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF 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. - - -class MetaData(object): - def __init__(self): - self.data = {} - - def addMetaData(self, key, value): - if key not in self.data: - self.data[key] = [] - self.data[key].append(value) - - def toXmlRpc(self): - return self.data http://git-wip-us.apache.org/repos/asf/oodt/blob/d329e1ab/workflow/src/main/python/xmlrpc-wrapper/oodt_workflow_xmlrpc.py ---------------------------------------------------------------------- diff --git a/workflow/src/main/python/xmlrpc-wrapper/oodt_workflow_xmlrpc.py b/workflow/src/main/python/xmlrpc-wrapper/oodt_workflow_xmlrpc.py deleted file mode 100644 index ae83c23..0000000 --- a/workflow/src/main/python/xmlrpc-wrapper/oodt_workflow_xmlrpc.py +++ /dev/null @@ -1,134 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF 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. - - -import xmlrpclib -from oodt_metadata import MetaData - - -class OODTWorkFlowWrapper(object): - def __init__(self, host): - self._server = xmlrpclib.Server(host) - - #String executeDynamicWorkflow(Vector<String> taskIds, Hashtable metadata) - def executeDynamicWorkflow(self, taskIds, metadata): - return self._server.workflowmgr.executeDynamicWorkflow(taskIds, metadata) - #Hashtable getConditionById(String conditionId) - def getConditionById(self, conditionId): - return self._server.workflowmgr.getConditionById(conditionId) - #Hashtable getFirstPage() - def getFirstPage(self): - return self._server.workflowmgr.getFirstPage() - #Hashtable getLastPage() - def getLastPage(self): - return self._server.workflowmgr.getLastPage() - #Hashtable getNextPage(Hashtable currentPage) - def getNextPage(self, currentPage): - return self._server.workflowmgr.getNextPage() - #int getNumWorkflowInstances() - def getNumWorkflowInstances(self): - return self._server.workflowmgr.getNumWorkflowInstances() - #int getNumWorkflowInstancesByStatus(String status) - def getNumWorkflowInstancesByStatus(self, status): - return self._server.workflowmgr.getNumWorkflowInstancesByStatus(status) - #Hashtable getPrevPage(Hashtable currentPage) - def getPrevPage(self, currentPage): - return self._server.workflowmgr.getPrevPage(currentPage) - #Vector getRegisteredEvents() - def getEventNames(self): - return self._server.workflowmgr.getRegisteredEvents() - #Hashtable getTaskById(String taskId) - def getTaskById(self, taskId): - return self._server.workflowmgr.getTaskById(taskId) - #Hashtable getWorkflowById(String workflowId) - def getWorkflowById(self, workflowId): - return self._server.workflowmgr.getWorkflowById(workflowId) - #double getWorkflowCurrentTaskWallClockMinutes(String workflowInstId) - def getWorkflowCurrentTaskWallClockMinutes(self, workflowInstId): - return self._server.workflowmgr.getWorkflowCurrentTaskWallClockMinutes(workflowInstId) - #Hashtable getWorkflowInstanceById(String wInstId) - def getWorkflowInstanceById(self,wInstId): - return self._server.workflowmgr.getWorkflowInstanceById(wInstId) - #Hashtable getWorkflowInstanceMetadata(String wInstId) - def getWorkflowInstanceMetadata(self, wInstId): - return self._server.workflowmgr.getWorkflowInstanceMetadata(wInstId) - #Vector getWorkflowInstances() - def getWorkflowInstances(self): - return self._server.workflowmgr.getWorkflowInstances() - #Vector getWorkflowInstancesByStatus(String status) - def getWorkflowInstancesByStatus(self, status): - return self._server.workflowmgr.getWorkflowInstancesByStatus(status) - #Vector getWorkflows() - def getWorkflows(self): - return self._server.workflowmgr.getWorkflows() - #Vector getWorkflowsByEvent(String eventName) - def getWorkflowsByEvent(self, eventName): - return self._server.workflowmgr.getWorkflowsByEvent(eventName) - #double getWorkflowWallClockMinutes(String workflowInstId) - def getWorkflowWallClockMinutes(self,workflowInstId): - return getWorkflowWallClockMinutes(workflowInstId) - #boolean handleEvent(String eventName, Hashtable metadata) - def startEvent(self, eventName, metaData): - return self._server.workflowmgr.handleEvent(eventName, metaData) - #Hashtable paginateWorkflowInstances(int pageNum) - def paginateWorkflowInstances(self, pageNum): - return self._server.workflowmgr.paginateWorkflowInstances(pageNum) - #Hashtable paginateWorkflowInstances(int pageNum, String status)(self, workflowInstId) - def paginateWorkflowInstances(self, pageNum, status): - return self._server.workflowmgr.paginateWorkflowInstances(pageNum, status) - #boolean pauseWorkflowInstance(String workflowInstId) - def pauseWorkflowInstance(self, workflowInstId): - return self._server.workflowmgr.pauseWorkflowInstance(workflowInstId) - #boolean resumeWorkflowInstance(String workflowInstId) - def resumeWorkflowInstance(self, workflowInstId): - return self._server.workflowmgr.resumeWorkflowInstance(workflowInstId) - #boolean setWorkflowInstanceCurrentTaskEndDateTime(String wInstId, String endDateTimeIsoStr) - def setWorkflowInstanceCurrentTaskEndDateTime(self, workflowInstId, endDateTimeIsoStr): - return self._server.workflowmgr.setWorkflowInstanceCurrentTaskEndDateTime(workflowInstId, endDateTimeIsoStr) - #boolean setWorkflowInstanceCurrentTaskStartDateTime(String wInstId, String startDateTimeIsoStr) - def setWorkflowInstanceCurrentTaskStartDateTime(self, workflowInstId, startDateTimeIsoStr): - return self._server.workflowmgr.setWorkflowInstanceCurrentTaskStartDateTime(workflowInstId, startDateTimeIsoStr) - #boolean stopWorkflowInstance(String workflowInstId) - def stopWorkflowInstance(self, workflowInstId): - return self._server.workflowmgr.stopWorkflowInstance(workflowInstId) - #boolean updateMetadataForWorkflow(String workflowInstId, Hashtable metadata) - def updateMetadataForWorkflow(self, workflowInstId, metadata): - return self._server.workflowmgr.updateMetadataForWorkflow(workflowInstId, metadata) - #boolean updateWorkflowInstance(Hashtable workflowInst) - def updateWorkflowInstance(workflowInst): - return self._server.workflowmgr.updateWorkflowInstance(workflowInst) - #boolean updateWorkflowInstanceStatus(String workflowInstanceId, String status) - def updateWorkflowInstanceStatus(self, workflowInstId, status): - return self._server.workflowmgr.updateWorkflowInstanceStatus(workflowInstId, status) - - - -def main(): - # create instance - oodt = OODTWorkFlowWrapper("http://localhost:9200";) - # get event info - events = oodt.getEventNames() - # create metadata object to invoke an event - met = MetaData() - met.addMetaData("hello", "world") - # print available events - print 'available events:', events - - # oodt.startEvent(events[0], met.toXmlRpc()) - -if __name__ == '__main__': - main()
