diff --git a/conf/shiro.ini.oauth b/conf/shiro.ini.oauth new file mode 100644 index 00000000000..8ecf2a3aec5 --- /dev/null +++ b/conf/shiro.ini.oauth @@ -0,0 +1,164 @@ +# +# 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. +# + +# Sample LDAP configuration, for user Authentication, currently tested for single Realm +[main] +#roleAdminAuthGenerator = org.pac4j.http.authorization.generator.RememberMeAuthorizationGenerator + +facebookClient = org.pac4j.oauth.client.FacebookClient +facebookClient.key = +facebookClient.secret = +facebookClient.callbackUrl = http://localhost:8080 +facebookClient.scope = email + +oauthClient = org.pac4j.oauth.client.GenericOAuth20Client +oauthClient.key = acme +oauthClient.secret = acmesecret +oauthClient.authUrl = http://localhost:8180/app/v2/user/login +oauthClient.tokenUrl = http://localhost:8180/oauth/token +oauthClient.callbackUrl = http://localhost:8080 +#oauthClient.authorizationGenerator = $roleAdminAuthGenerator + +#clients.callbackUrl = http://localhost:8080 +clients.clients = $oauthClient +#clients.clients = $facebookClient + +#requireRoleAdmin = org.pac4j.core.authorization.authorizer.RequireAnyRoleAuthorizer +#requireRoleAdmin.elements = ROLE_ADMIN +#requireRoleAdmin.elements = admin + +#excludedPathMatcher = org.pac4j.core.matching.PathMatcher +#excludedPathMatcher.excludedPath = /facebook/notprotected.jsp + +#config.authorizers = admin:$requireRoleAdmin +#config.matchers = excludedPath:$excludedPathMatcher + +callbackFilter.defaultUrl = / + +authc = io.buji.pac4j.filter.SecurityFilter +authc.config = $config +authc.clients = GenericOAuth20Client +#authc.clients = FacebookClient + + +pac4jRealm = io.buji.pac4j.realm.Pac4jRealm +pac4jSubjectFactory = io.buji.pac4j.subject.Pac4jSubjectFactory +securityManager.subjectFactory = $pac4jSubjectFactory +securityManager.realms = $pac4jRealm + +#sessionManager = org.apache.shiro.web.session.mgt.DefaultWebSessionManager +#securityManager.sessionManager = $sessionManager +#securityManager.sessionManager.globalSessionTimeout = 86400000 +#cacheManager = org.apache.shiro.cache.MemoryConstrainedCacheManager +#securityManager.cacheManager = $cacheManager + +# ----------------------- + +### A sample for configuring Active Directory Realm +#activeDirectoryRealm = org.apache.zeppelin.realm.ActiveDirectoryGroupRealm +#activeDirectoryRealm.systemUsername = userNameA + +#use either systemPassword or hadoopSecurityCredentialPath, more details in http://zeppelin.apache.org/docs/latest/security/shiroauthentication.html +#activeDirectoryRealm.systemPassword = passwordA +#activeDirectoryRealm.hadoopSecurityCredentialPath = jceks://file/user/zeppelin/zeppelin.jceks +#activeDirectoryRealm.searchBase = CN=Users,DC=SOME_GROUP,DC=COMPANY,DC=COM +#activeDirectoryRealm.url = ldap://ldap.test.com:389 +#activeDirectoryRealm.groupRolesMap = "CN=admin,OU=groups,DC=SOME_GROUP,DC=COMPANY,DC=COM":"admin","CN=finance,OU=groups,DC=SOME_GROUP,DC=COMPANY,DC=COM":"finance","CN=hr,OU=groups,DC=SOME_GROUP,DC=COMPANY,DC=COM":"hr" +#activeDirectoryRealm.authorizationCachingEnabled = false + +### A sample for configuring LDAP Directory Realm +#ldapRealm = org.apache.zeppelin.realm.LdapGroupRealm +## search base for ldap groups (only relevant for LdapGroupRealm): +#ldapRealm.contextFactory.environment[ldap.searchBase] = dc=COMPANY,dc=COM +#ldapRealm.contextFactory.url = ldap://ldap.test.com:389 +#ldapRealm.userDnTemplate = uid={0},ou=Users,dc=COMPANY,dc=COM +#ldapRealm.contextFactory.authenticationMechanism = simple + +### A sample PAM configuration +#pamRealm=org.apache.zeppelin.realm.PamRealm +#pamRealm.service=sshd + +### A sample for configuring ZeppelinHub Realm +#zeppelinHubRealm = org.apache.zeppelin.realm.ZeppelinHubRealm +## Url of ZeppelinHub +#zeppelinHubRealm.zeppelinhubUrl = https://www.zeppelinhub.com +#securityManager.realms = $zeppelinHubRealm + +## A same for configuring Knox SSO Realm +#knoxJwtRealm = org.apache.zeppelin.realm.jwt.KnoxJwtRealm +#knoxJwtRealm.providerUrl = https://domain.example.com/ +#knoxJwtRealm.login = gateway/knoxsso/knoxauth/login.html +#knoxJwtRealm.logout = gateway/knoxssout/api/v1/webssout +#knoxJwtRealm.logoutAPI = true +#knoxJwtRealm.redirectParam = originalUrl +#knoxJwtRealm.cookieName = hadoop-jwt +#knoxJwtRealm.publicKeyPath = /etc/zeppelin/conf/knox-sso.pem +# +#knoxJwtRealm.groupPrincipalMapping = group.principal.mapping +#knoxJwtRealm.principalMapping = principal.mapping +#authc = org.apache.zeppelin.realm.jwt.KnoxAuthenticationFilter + +#sessionManager = org.apache.shiro.web.session.mgt.DefaultWebSessionManager + +### If caching of user is required then uncomment below lines +#cacheManager = org.apache.shiro.cache.MemoryConstrainedCacheManager +#securityManager.cacheManager = $cacheManager + +### Enables 'HttpOnly' flag in Zeppelin cookies +#cookie = org.apache.shiro.web.servlet.SimpleCookie +#cookie.name = JSESSIONID +#cookie.httpOnly = true +### Uncomment the below line only when Zeppelin is running over HTTPS +#cookie.secure = true +#sessionManager.sessionIdCookie = $cookie + +#securityManager.sessionManager = $sessionManager +# 86,400,000 milliseconds = 24 hour +#securityManager.sessionManager.globalSessionTimeout = 86400000 +#shiro.loginUrl = /api/login + +#[roles] +#role1 = * +#role2 = * +#role3 = * +#admin = * + +[urls] +# This section is used for url-based security. For details see the shiro.ini documentation. +# +# You can secure interpreter, configuration and credential information by urls. +# Comment or uncomment the below urls that you want to hide: +# anon means the access is anonymous. +# authc means form based auth Security. +# +# IMPORTANT: Order matters: URL path expressions are evaluated against an incoming request +# in the order they are defined and the FIRST MATCH WINS. +# +# To allow anonymous access to all but the stated urls, +# uncomment the line second last line (/** = anon) and comment the last line (/** = authc) +# +/api/version = anon +# Allow all authenticated users to restart interpreters on a notebook page. +# Comment out the following line if you would like to authorize only admin users to restart interpreters. +#/api/interpreter/setting/restart/** = authc +#/api/interpreter/** = authc, roles[admin] +#/api/notebook-repositories/** = authc, roles[admin] +#/api/configurations/** = authc, roles[admin] +#/api/credential/** = authc, roles[admin] +#/api/admin/** = authc, roles[admin] +#/** = anon +/** = authc diff --git a/conf/shiro.ini.template b/conf/shiro.ini.template index 23aa473a7f0..978b69c0330 100644 --- a/conf/shiro.ini.template +++ b/conf/shiro.ini.template @@ -56,6 +56,11 @@ user3 = password4, role2 #zeppelinHubRealm.zeppelinhubUrl = https://www.zeppelinhub.com #securityManager.realms = $zeppelinHubRealm +### A sample for configuring Metatron Realm +#metatronRealm = org.apache.zeppelin.realm.MetatronRealm +#metatronRealm.authUrl = http://localhost:8180/oauth/token +#securityManager.realms = $metatronRealm + ## A same for configuring Knox SSO Realm #knoxJwtRealm = org.apache.zeppelin.realm.jwt.KnoxJwtRealm #knoxJwtRealm.providerUrl = https://domain.example.com/ diff --git a/metatron/pom.xml b/metatron/pom.xml new file mode 100644 index 00000000000..91fc963f954 --- /dev/null +++ b/metatron/pom.xml @@ -0,0 +1,133 @@ + + + + + 4.0.0 + + + zeppelin-interpreter-parent + org.apache.zeppelin + 0.9.0-SNAPSHOT + ../zeppelin-interpreter-parent/pom.xml + + + zeppelin-metatron + jar + 0.9.0-SNAPSHOT + Zeppelin: Metatron interpreter + + + metatron + 4.0.2 + 18.0 + 0.1.6 + 1.4.9 + + + + + com.google.code.gson + gson + + + + commons-lang + commons-lang + + + + org.apache.httpcomponents + httpasyncclient + ${httpasyncclient.version} + + + + com.google.guava + guava + ${guava.version} + + + + com.github.wnameless + json-flattener + ${json-flattener.version} + + + + org.antlr + antlr4-runtime + 4.7.2 + + + + com.mashape.unirest + unirest-java + ${unirest.version} + + + org.apache.zeppelin + zeppelin-zengine + 0.9.0-SNAPSHOT + + + + + + + + maven-enforcer-plugin + + + maven-dependency-plugin + + + maven-resources-plugin + + + maven-shade-plugin + + + org.apache.maven.plugins + maven-checkstyle-plugin + + true + + + + org.antlr + antlr4-maven-plugin + 4.7 + + + -package + org.apache.zeppelin.metatron.antlr + + + + + antlr + + antlr4 + + + + + + + + diff --git a/metatron/src/main/antlr4/Metatron.g4 b/metatron/src/main/antlr4/Metatron.g4 new file mode 100644 index 00000000000..b694a8efdb4 --- /dev/null +++ b/metatron/src/main/antlr4/Metatron.g4 @@ -0,0 +1,11 @@ +grammar Metatron; +exprs: (stmt TERMINATOR)* | stmt ; + +stmt: 'describe' RESOURCE ; + +RESOURCE : [A-Za-z0-9_-]+ ; + +TERMINATOR : ';' ; + +WS: [ \n\t\r]+ -> skip; +IDENTIFIER : [a-zA-Z0-9_-]+; \ No newline at end of file diff --git a/metatron/src/main/java/org/apache/zeppelin/metatron/MetatronInterpreter.java b/metatron/src/main/java/org/apache/zeppelin/metatron/MetatronInterpreter.java new file mode 100644 index 00000000000..cd68e4030db --- /dev/null +++ b/metatron/src/main/java/org/apache/zeppelin/metatron/MetatronInterpreter.java @@ -0,0 +1,415 @@ +/* + * 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. + */ + +package org.apache.zeppelin.metatron; + +import com.google.common.collect.ImmutableList; +import com.google.gson.Gson; +import com.google.gson.GsonBuilder; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.util.Collection; +import java.util.LinkedList; +import java.util.Iterator; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +import org.antlr.v4.runtime.CharStreams; +import org.antlr.v4.runtime.CommonTokenStream; +import org.apache.zeppelin.interpreter.InterpreterException; +import org.apache.zeppelin.interpreter.InterpreterResultMessage; +import org.apache.zeppelin.metatron.antlr.MetatronLexer; +import org.apache.zeppelin.metatron.antlr.MetatronParser; +import org.apache.zeppelin.metatron.client.MetatronClient; +import org.apache.zeppelin.metatron.message.*; +import org.apache.zeppelin.user.AuthenticationInfo; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; +import java.util.stream.Collectors; + +import org.apache.zeppelin.interpreter.Interpreter; +import org.apache.zeppelin.interpreter.InterpreterContext; +import org.apache.zeppelin.interpreter.InterpreterResult; +import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion; + +/** + * Metatron Interpreter for Zeppelin. + */ +public class MetatronInterpreter extends Interpreter { + private static Logger LOGGER = LoggerFactory.getLogger(MetatronInterpreter.class); + + public static final String METATRON_URL = "metatron.url"; + + private final Gson gson = new GsonBuilder().setPrettyPrinting().create(); + private final Pattern showDatabasesPattern; + private final Pattern showDetailPattern; + private final Pattern getDataPattern; + private final Pattern sqlQueryPattern; + private MetatronClient client; + + public MetatronInterpreter(Properties property) { + super(property); + showDatabasesPattern = Pattern.compile("show datasources"); + showDetailPattern = Pattern.compile("show (?.*)"); + getDataPattern = Pattern.compile("datasource=(?[^ ]+)[ ](?[^ ]+)[ ]limit=(?[0-9]+)[ ](?[^ ]+)[ ](?[^ ]+)"); + sqlQueryPattern = Pattern.compile("select (.*)"); + } + + @Override + public void open() { + try { + client = new MetatronClient(getProperty(METATRON_URL)); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + @Override + public void close() { + } + + @Override + public InterpreterResult interpret(String cmd, InterpreterContext interpreterContext) { + if (interpreterContext != null) { + interpreterContext.getResourcePool().put("metatron", this); + client.setAccessToken(interpreterContext + .getAuthenticationInfo() + .getUserCredentials() + .getUsernamePassword("token") + .getPassword()); + } + + try { + InterpreterResult result = runMetatronQuery(cmd, interpreterContext); + if (result != null) { + return result; + } else { + return new InterpreterResult(InterpreterResult.Code.ERROR, String.format("Unknown expression '%s'", cmd)); + } + } catch (IOException e) { + return new InterpreterResult(InterpreterResult.Code.ERROR, e.getMessage()); + } + } + + public InterpreterResult runMetatronQuery(String query) throws IOException { + return runMetatronQuery(query, null); + } + + String ignoreComment(String input){ + + String[] change_target = input.split("\\n"); + + StringBuilder result = new StringBuilder(); + + String prefix = ""; + for( String curLine : change_target ){ + + String tLine = curLine.trim(); + + if( tLine.length() > 0 && tLine.charAt(0) == '#'){ + continue; + } + + result.append(prefix); + prefix = "\n"; + result.append(curLine); + } + + return result.toString(); + + } + + InterpreterResult runMetatronQuery(String query, InterpreterContext interpreterContext) throws IOException { + + + query = ignoreComment(query); + + Matcher m = showDatabasesPattern.matcher(query); + if (m.matches()) { + List resp = client.showDatasources(); + return new InterpreterResult( + InterpreterResult.Code.SUCCESS, + ImmutableList.of( + datasourcesToTable(resp) + ) + ); + } + + m = showDetailPattern.matcher(query); + if (m.matches()) { + DatasourceDetail detail = client.showDatasource(m.group("datasource")); + + StringBuilder summary = new StringBuilder(); + String summaryFormat = "%-20s: %s\n"; + summary.append(String.format(summaryFormat, "Created by", detail.getCreatedBy().getFullName())); + summary.append(String.format(summaryFormat, "Published", detail.isPublished())); + summary.append(String.format(summaryFormat, "Status", detail.getStatus())); + summary.append(String.format(summaryFormat, "Description", detail.getDescription())); + + StringBuilder fields = new StringBuilder(); + fields.append("id\tname\tlogicalName\ttype\tlogicalType\trole\taggrType\tseq\n"); + for (Field f : detail.getFields()) { + fields.append(String.valueOf(f.getId()) + '\t' + + f.getName() + '\t' + + f.getLogicalName() + '\t' + + f.getType() + '\t' + + f.getLogicalType() + '\t' + + f.getRole() + '\t' + + f.getAggrType() + '\t' + + String.valueOf(f.getSeq()) + '\n'); + } + + return new InterpreterResult( + InterpreterResult.Code.SUCCESS, + ImmutableList.of( + new InterpreterResultMessage( + InterpreterResult.Type.TEXT, summary.toString()), + new InterpreterResultMessage( + InterpreterResult.Type.TABLE, fields.toString()) + ) + ); + } + + m = getDataPattern.matcher(query); + if (m.matches()) { + String datasourceName = m.group("datasource"); + String filterExpr = m.group("filter"); + String limit = m.group("limit"); + String dimension = m.group("dimension"); + String measure = m.group("measure"); + + List filters = new LinkedList<>(); + for (String expr : filterExpr.split(",")) { + String[] fieldValue = expr.split("="); + filters.add(Filter.newBuilder() + .setType("include") + .setField(fieldValue[0]) + .addValue(fieldValue[1]) + .build()); + } + + DataResponse data = client.getData( + datasourceName, + filters, + ImmutableList.of( + new Projection("dimension", dimension), + new Projection("measure", measure)), + new Limits(Long.parseLong(limit)) + ); + + StringBuilder table = new StringBuilder(); + + if (interpreterContext != null) { + interpreterContext.getResourcePool().put("data", data); + } + + // create header + if (data.size() <= 0) { + return new InterpreterResult(InterpreterResult.Code.SUCCESS); + } + + for (String key : data.get(0).keySet()) { + if (table.toString().length() > 0) { + table.append("\t"); + } + table.append(key); + } + table.append("\n"); + + // add rows + for (Record r : data) { + Collection values = r.values(); + int i = 0; + for (Object v : values) { + if (i++ > 0) { + table.append("\t"); + } + table.append(v); + } + table.append("\n"); + } + + return new InterpreterResult(InterpreterResult.Code.SUCCESS, InterpreterResult.Type.TABLE, table.toString()); + } + + + m = sqlQueryPattern.matcher(query); + if (m.matches()) { + + SQLQueryResponse sqlQueryResponse = client.runSQLQuery( query ); + + + + StringBuilder table = new StringBuilder(); + + if (interpreterContext != null) { + interpreterContext.getResourcePool().put("data", sqlQueryResponse); + } + + // create header + if (sqlQueryResponse.getData().size() <= 0) { + return new InterpreterResult(InterpreterResult.Code.SUCCESS); + } + + + for (Record r : sqlQueryResponse.getFields()) { + if (table.toString().length() > 0) { + table.append("\t"); + } + table.append(r.get("name")); + } + table.append("\n"); + + + // add rows + for (Record r : sqlQueryResponse.getData()) { + + Iterator i = r.values().iterator(); + + if( i.hasNext() ) { + table.append(i.next()); + } + + while( i.hasNext()) { + table.append("\t"); + table.append(i.next()); + } + + table.append("\n"); + } + + return new InterpreterResult(InterpreterResult.Code.SUCCESS, InterpreterResult.Type.TABLE, table.toString()); + } + + + + return null; +// // parse statements using antlr and execute +// MetatronParser parser = parseMetatronExpr(query); +// List results = execStatement(parser.exprs().stmt()); +// if (results != null && results.size() > 0) { +// return new InterpreterResult(InterpreterResult.Code.SUCCESS, results); +// } else { +// return null; +// } + } + + MetatronParser parseMetatronExpr(String cmd) throws IOException { + InputStream inputStream = new ByteArrayInputStream(cmd.getBytes(StandardCharsets.UTF_8)); + MetatronLexer lexer = new org.apache.zeppelin.metatron.antlr.MetatronLexer(CharStreams.fromStream(inputStream, StandardCharsets.UTF_8)); + MetatronParser parser = new org.apache.zeppelin.metatron.antlr.MetatronParser(new CommonTokenStream(lexer)); + return parser; + } + +// List execStatement(List stmts) { +// return stmts.stream() +// .flatMap(s -> execStatement(s).stream()) +// .collect(Collectors.toList()); +// } +// +// List execStatement(MetatronParser.StmtContext stmt) { +//// String resource = stmt.RESOURCE().getText(); +// String resource = "sales"; +// +// DatasourceDetail detail = null; +// try { +// detail = client.showDatasource(resource); +// } catch (IOException e) { +// logger.error("Can't get datasource detail " + resource); +// return ImmutableList.of( +// new InterpreterResultMessage(InterpreterResult.Type.TEXT, e.getMessage()) +// ); +// } +// +// StringBuilder summary = new StringBuilder(); +// String summaryFormat = "%-20s: %s\n"; +// summary.append(String.format(summaryFormat, "Created by", detail.getCreatedBy().getFullName())); +// summary.append(String.format(summaryFormat, "Published", detail.isPublished())); +// summary.append(String.format(summaryFormat, "Status", detail.getStatus())); +// summary.append(String.format(summaryFormat, "Description", detail.getDescription())); +// +// StringBuilder fields = new StringBuilder(); +// fields.append("id\tname\talias\ttype\tlogicalType\trole\tbiType\n"); +// for (Field f : detail.getFields()) { +// fields.append(f.getId() + '\t' + +// f.getName() + '\t' + +// f.getAlias() + '\t' + +// f.getType() + '\t' + +// f.getLogicalType() + '\t' + +// f.getRole() + '\t' + +// f.getBiType() + '\n'); +// } +// +// return ImmutableList.of( +// new InterpreterResultMessage( +// InterpreterResult.Type.TEXT, summary.toString()), +// new InterpreterResultMessage( +// InterpreterResult.Type.TABLE, fields.toString()) +// ); +// } + + private InterpreterResultMessage datasourcesToTable(List ds) { + StringBuilder table = new StringBuilder(); + table.append("id\tname\ttype\tengine\tdescription\n"); + + for(Datasource d : ds) { + table.append(d.getId() + '\t' + d.getName() + '\t' + d.getConnType() + '\t' + d.getEngineName() + '\t' + d.getDescription() + '\n') ; + } + return new InterpreterResultMessage(InterpreterResult.Type.TABLE, table.toString()); + } + + + @Override + public void cancel(InterpreterContext interpreterContext) { + // Nothing to do + } + + @Override + public FormType getFormType() { + return FormType.SIMPLE; + } + + @Override + public int getProgress(InterpreterContext interpreterContext) { + return 0; + } + + @Override + public List completion(String s, int i, + InterpreterContext interpreterContext) { + final List suggestions = new ArrayList<>(); + return suggestions; + } + + public MetatronClient getClient() { + return client; + } + + private String getTokenForApiRequest(InterpreterContext interpreterContext) { + AuthenticationInfo authenticationInfo = interpreterContext.getAuthenticationInfo(); + // TODO 'authentication' should hold token from Metatron SSO + // For now, request token manually here + return null; + } +} diff --git a/metatron/src/main/java/org/apache/zeppelin/metatron/client/DateDeserializer.java b/metatron/src/main/java/org/apache/zeppelin/metatron/client/DateDeserializer.java new file mode 100644 index 00000000000..5bbecfbc35c --- /dev/null +++ b/metatron/src/main/java/org/apache/zeppelin/metatron/client/DateDeserializer.java @@ -0,0 +1,31 @@ +package org.apache.zeppelin.metatron.client; + +import com.google.gson.JsonParseException; +import java.lang.reflect.Type; +import com.google.gson.JsonDeserializationContext; +import com.google.gson.JsonDeserializer; +import com.google.gson.JsonElement; +import java.text.ParseException; +import java.text.SimpleDateFormat; +import java.util.Date; +import java.util.TimeZone; + +public class DateDeserializer implements JsonDeserializer { + SimpleDateFormat formatter; + + public DateDeserializer() { + formatter = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"); + formatter.setTimeZone(TimeZone.getTimeZone("UTC")); + } + + @Override + public Date deserialize(JsonElement element, Type arg1, JsonDeserializationContext arg2) throws JsonParseException { + String date = element.getAsString(); + + try { + return formatter.parse(date); + } catch (ParseException e) { + throw new JsonParseException(e); + } + } +} diff --git a/metatron/src/main/java/org/apache/zeppelin/metatron/client/MetatronClient.java b/metatron/src/main/java/org/apache/zeppelin/metatron/client/MetatronClient.java new file mode 100644 index 00000000000..823ee2dbf3d --- /dev/null +++ b/metatron/src/main/java/org/apache/zeppelin/metatron/client/MetatronClient.java @@ -0,0 +1,215 @@ +package org.apache.zeppelin.metatron.client; + +import com.google.common.collect.ImmutableList; +import com.google.gson.Gson; +import com.google.gson.GsonBuilder; +import com.google.gson.reflect.TypeToken; +import java.io.IOException; +import java.io.UnsupportedEncodingException; +import java.net.URL; +import java.util.Date; +import java.util.List; +import org.apache.commons.httpclient.HttpClient; +import org.apache.commons.io.Charsets; +import org.apache.commons.io.IOUtils; +import org.apache.http.HttpResponse; +import org.apache.http.client.entity.UrlEncodedFormEntity; +import org.apache.http.client.methods.CloseableHttpResponse; +import org.apache.http.client.methods.HttpGet; +import org.apache.http.client.methods.HttpPost; +import org.apache.http.client.methods.HttpUriRequest; +import org.apache.http.entity.StringEntity; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.client.HttpClients; +import org.apache.http.message.BasicNameValuePair; +import org.apache.zeppelin.metatron.message.*; + +/** + * Metatron Http client + */ +public class MetatronClient { + + private final String baseUrl; + private Gson gson; + private String accessToken; + CloseableHttpClient httpClient = HttpClients.createDefault(); + + enum RequestPath { + OAUTH_TOKEN("/oauth/token"), + API_DATASOURCES("/api/datasources"), + API_DATASOURCES_QUERY("/api/datasources/query/search"), + API_SQL_QUERY("/api/connections/query/data?extractColumnName=true&limit=100000"); + + String path; + RequestPath(String path) { + this.path = path; + } + + public String path() { + return path; + } + } + + public MetatronClient(String baseUrl) throws IOException { + this.baseUrl = baseUrl; + gson = new GsonBuilder() + .registerTypeAdapter(Date.class, new DateDeserializer()) + .create(); + } + + + + public List showDatasources() throws IOException { + String url = baseUrl + "/api/datasources"; + + HttpGet get = httpGet(RequestPath.API_DATASOURCES); + CloseableHttpResponse resp = httpClient.execute(get); + DatasourcesResponse dsResp = readResponse(resp, DatasourcesResponse.class); + return dsResp.getDatasources(); + } + + public DatasourceDetail showDatasource(String dsName) throws IOException { + Datasource ds = getDatasourceByName(dsName); + if (ds == null) { + return null; + } + + String url = String.format("%s/%s?projection=forDetailView", RequestPath.API_DATASOURCES.path(), ds.getId()); + + HttpGet get = httpGet(url); + CloseableHttpResponse resp = httpClient.execute(get); + return readResponse(resp, DatasourceDetail.class); + } + + private Datasource getDatasourceByName(String name) throws IOException { + List datasources = showDatasources(); + for (Datasource ds : datasources) { + if (name.equals(ds.getName())) { + return ds; + } + } + return null; + } + + public DataResponse getData( + String datasourceName, + List filters, + List projections, + Limits limits) throws IOException { + Datasource ds = getDatasourceByName(datasourceName); + if (ds == null) { + throw new IOException("Data source not found"); + } + + DataRequest dr = new DataRequest( + new DatasourceRequest(ds.getEngineName(), "default", false), + filters, + projections, + limits, + true + ); + + return getData(dr); + } + + public DataResponse getData(DataRequest dataRequest) throws IOException { + HttpPost post = httpPost(RequestPath.API_DATASOURCES_QUERY); + try { + String request = gson.toJson(dataRequest); + post.setEntity(new StringEntity(request)); + } catch (UnsupportedEncodingException e) { + throw new IOException(e); + } + + CloseableHttpResponse resp = httpClient.execute(post); + return readResponse(resp, DataResponse.class); + } + + + public SQLQueryResponse runSQLQuery( String strSQLQuery ) throws IOException { + + SQLQuery sqlQuery = new SQLQuery( strSQLQuery ); + + return runSQLQuery(sqlQuery); + } + + public SQLQueryResponse runSQLQuery(SQLQuery sqlQuery) throws IOException { + HttpPost post = httpPost(RequestPath.API_SQL_QUERY); + try { + String request = gson.toJson(sqlQuery); + post.setEntity(new StringEntity(request)); + } catch (UnsupportedEncodingException e) { + throw new IOException(e); + } + + CloseableHttpResponse resp = httpClient.execute(post); + return readResponse(resp, SQLQueryResponse.class); + } + + /** + * Get authentication token + * TODO: delete this method after after proper auth integration + * @return + */ + /* + AuthResponse auth() throws IOException { + int protoPos = baseUrl.indexOf("://"); + String url = ""; + + if (protoPos > 0) { + protoPos += "://".length(); + url = baseUrl.substring(0, protoPos); + } + url += "polaris_trusted:secret@" + baseUrl.substring(protoPos) + RequestPath.OAUTH_TOKEN.path(); + + HttpPost post = new HttpPost(url); + + post.setEntity(new UrlEncodedFormEntity(ImmutableList.of( + new BasicNameValuePair("grant_type", "client_credentials") + ))); + CloseableHttpResponse resp = httpClient.execute(post); + + String result = IOUtils.toString(resp.getEntity().getContent(), Charsets.UTF_8); + AuthResponse authResp = gson.fromJson(result, AuthResponse.class); + accessToken = authResp.getAccess_token(); + return authResp; + } + */ + + public void setAccessToken(String accessToken) { + this.accessToken = accessToken; + } + + HttpGet httpGet(RequestPath path) { + return httpGet(path.path()); + } + + HttpGet httpGet(String path) { + String url = baseUrl + path; + HttpGet get = new HttpGet(url); + setToken(get); + return get; + } + + HttpPost httpPost(RequestPath path) { + return httpPost(path.path()); + } + + HttpPost httpPost(String path) { + String url = baseUrl + path; + HttpPost post = new HttpPost(url); + setToken(post); + return post; + } + + T readResponse(HttpResponse resp, Class type) throws IOException { + String result = IOUtils.toString(resp.getEntity().getContent(), Charsets.UTF_8); + return gson.fromJson(result, type); + } + + void setToken(HttpUriRequest req) { + req.setHeader("Content-Type", "application/json"); + req.setHeader("Authorization", "bearer " + accessToken); + } + +} diff --git a/metatron/src/main/java/org/apache/zeppelin/metatron/message/AuthResponse.java b/metatron/src/main/java/org/apache/zeppelin/metatron/message/AuthResponse.java new file mode 100644 index 00000000000..42ac0f72725 --- /dev/null +++ b/metatron/src/main/java/org/apache/zeppelin/metatron/message/AuthResponse.java @@ -0,0 +1,29 @@ +package org.apache.zeppelin.metatron.message; + +public class AuthResponse { + String access_token; + String token_type; + long expires_in; + String scope; + String jti; + + public String getAccess_token() { + return access_token; + } + + public String getToken_type() { + return token_type; + } + + public long getExpires_in() { + return expires_in; + } + + public String getScope() { + return scope; + } + + public String getJti() { + return jti; + } +} diff --git a/metatron/src/main/java/org/apache/zeppelin/metatron/message/DataRequest.java b/metatron/src/main/java/org/apache/zeppelin/metatron/message/DataRequest.java new file mode 100644 index 00000000000..755de586c08 --- /dev/null +++ b/metatron/src/main/java/org/apache/zeppelin/metatron/message/DataRequest.java @@ -0,0 +1,46 @@ +package org.apache.zeppelin.metatron.message; + +import java.util.List; + +public class DataRequest { + + private DatasourceRequest dataSource; + private List filters; + private Limits limits; + private List projections; + private boolean preview; + + public DataRequest( + DatasourceRequest dataSource, + List filters, + List projections, + Limits limits, + boolean preview) + { + this.dataSource = dataSource; + this.filters = filters; + this.projections = projections; + this.limits = limits; + this.preview = preview; + } + + public DatasourceRequest getDataSource() { + return dataSource; + } + + public List getFilters() { + return filters; + } + + public Limits getLimits() { + return limits; + } + + public List getProjections() { + return projections; + } + + public boolean isPreview() { + return preview; + } +} diff --git a/metatron/src/main/java/org/apache/zeppelin/metatron/message/DataResponse.java b/metatron/src/main/java/org/apache/zeppelin/metatron/message/DataResponse.java new file mode 100644 index 00000000000..5761a693705 --- /dev/null +++ b/metatron/src/main/java/org/apache/zeppelin/metatron/message/DataResponse.java @@ -0,0 +1,6 @@ +package org.apache.zeppelin.metatron.message; + +import java.util.LinkedList; + +public class DataResponse extends LinkedList { +} diff --git a/metatron/src/main/java/org/apache/zeppelin/metatron/message/Datasource.java b/metatron/src/main/java/org/apache/zeppelin/metatron/message/Datasource.java new file mode 100644 index 00000000000..af922e9a4b0 --- /dev/null +++ b/metatron/src/main/java/org/apache/zeppelin/metatron/message/Datasource.java @@ -0,0 +1,56 @@ +package org.apache.zeppelin.metatron.message; + +import java.util.Date; +import java.util.List; +import java.util.Map; + +public class Datasource { + String name; + String engineName; + String id; + String description; + String connType; + Date modifiedTime; + boolean published; + DatasourceSummary summary; + + List fields; + + Map _links; + + public String getName() { + return name; + } + + public String getEngineName() { + return engineName; + } + + public String getId() { + return id; + } + + public String getDescription() { + return description; + } + + public String getConnType() { + return connType; + } + + public Date getModifiedTime() { + return modifiedTime; + } + + public boolean isPublished() { + return published; + } + + public List getFields() { + return fields; + } + + public Map get_links() { + return _links; + } +} diff --git a/metatron/src/main/java/org/apache/zeppelin/metatron/message/DatasourceDetail.java b/metatron/src/main/java/org/apache/zeppelin/metatron/message/DatasourceDetail.java new file mode 100644 index 00000000000..6f218833846 --- /dev/null +++ b/metatron/src/main/java/org/apache/zeppelin/metatron/message/DatasourceDetail.java @@ -0,0 +1,41 @@ +package org.apache.zeppelin.metatron.message; + +import java.util.Map; + +public class DatasourceDetail extends Datasource { + String granularity; + Map contexts; + User createdBy; + User modifiedBy; + String status; + String dsType; + String srcType; + + public String getGranularity() { + return granularity; + } + + public Map getContexts() { + return contexts; + } + + public User getCreatedBy() { + return createdBy; + } + + public User getModifiedBy() { + return modifiedBy; + } + + public String getStatus() { + return status; + } + + public String getDsType() { + return dsType; + } + + public String getSrcType() { + return srcType; + } +} diff --git a/metatron/src/main/java/org/apache/zeppelin/metatron/message/DatasourceRequest.java b/metatron/src/main/java/org/apache/zeppelin/metatron/message/DatasourceRequest.java new file mode 100644 index 00000000000..d9b5e5179b9 --- /dev/null +++ b/metatron/src/main/java/org/apache/zeppelin/metatron/message/DatasourceRequest.java @@ -0,0 +1,34 @@ +package org.apache.zeppelin.metatron.message; + +import java.util.LinkedList; +import java.util.List; + +public class DatasourceRequest { + String name; + String type; + boolean temporary; + List joins; + + public DatasourceRequest(String name, String type, boolean temporary) { + this.name = name; + this.type = type; + this.temporary = temporary; + this.joins = new LinkedList<>(); + } + + public String getName() { + return name; + } + + public String getType() { + return type; + } + + public boolean isTemporary() { + return temporary; + } + + public List getJoins() { + return joins; + } +} diff --git a/metatron/src/main/java/org/apache/zeppelin/metatron/message/DatasourceSummary.java b/metatron/src/main/java/org/apache/zeppelin/metatron/message/DatasourceSummary.java new file mode 100644 index 00000000000..5f69c13b08d --- /dev/null +++ b/metatron/src/main/java/org/apache/zeppelin/metatron/message/DatasourceSummary.java @@ -0,0 +1,34 @@ +package org.apache.zeppelin.metatron.message; + +import java.util.Date; + +public class DatasourceSummary { + Date ingestionMinTime; + Date ingestionMaxTime; + Date lastAccessTime; + long size; + long count; + + public void DatasourceSummary() { + } + + public Date getIngestionMinTime() { + return ingestionMinTime; + } + + public Date getIngestionMaxTime() { + return ingestionMaxTime; + } + + public Date getLastAccessTime() { + return lastAccessTime; + } + + public long getSize() { + return size; + } + + public long getCount() { + return count; + } +} diff --git a/metatron/src/main/java/org/apache/zeppelin/metatron/message/DatasourcesResponse.java b/metatron/src/main/java/org/apache/zeppelin/metatron/message/DatasourcesResponse.java new file mode 100644 index 00000000000..c98f5811b87 --- /dev/null +++ b/metatron/src/main/java/org/apache/zeppelin/metatron/message/DatasourcesResponse.java @@ -0,0 +1,15 @@ +package org.apache.zeppelin.metatron.message; + +import java.util.List; + +public class DatasourcesResponse { + static class Embedded { + List datasources; + } + + Embedded _embedded; + + public List getDatasources() { + return _embedded.datasources; + } +} diff --git a/metatron/src/main/java/org/apache/zeppelin/metatron/message/Field.java b/metatron/src/main/java/org/apache/zeppelin/metatron/message/Field.java new file mode 100644 index 00000000000..8cd5cb0e9ca --- /dev/null +++ b/metatron/src/main/java/org/apache/zeppelin/metatron/message/Field.java @@ -0,0 +1,76 @@ +package org.apache.zeppelin.metatron.message; + +public class Field { + long id; + String name; + String logicalName; + String type; + String logicalType; + String role; + String aggrType; + long seq; + + public long getId() { + return id; + } + + public void setId(long id) { + this.id = id; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public String getLogicalName() { + return logicalName; + } + + public void setLogicalName(String logicalName) { + this.logicalName = logicalName; + } + + public String getType() { + return type; + } + + public void setType(String type) { + this.type = type; + } + + public String getLogicalType() { + return logicalType; + } + + public void setLogicalType(String logicalType) { + this.logicalType = logicalType; + } + + public String getRole() { + return role; + } + + public void setRole(String role) { + this.role = role; + } + + public String getAggrType() { + return aggrType; + } + + public void setAggrType(String aggrType) { + this.aggrType = aggrType; + } + + public long getSeq() { + return seq; + } + + public void setSeq(long seq) { + this.seq = seq; + } +} diff --git a/metatron/src/main/java/org/apache/zeppelin/metatron/message/Filter.java b/metatron/src/main/java/org/apache/zeppelin/metatron/message/Filter.java new file mode 100644 index 00000000000..f287e7b805c --- /dev/null +++ b/metatron/src/main/java/org/apache/zeppelin/metatron/message/Filter.java @@ -0,0 +1,57 @@ +package org.apache.zeppelin.metatron.message; + +import java.util.LinkedList; +import java.util.List; + +public class Filter { + String type; + String field; + List valueList; + + public Filter(String type, String field, List valueList) { + this.type = type; + this.field = field; + this.valueList = valueList; + } + + public static Builder newBuilder() { + return new Builder(); + } + + public static class Builder { + String type; + String field; + List valueList = new LinkedList<>(); + + public Builder setType(String type) { + this.type = type; + return this; + } + + public Builder setField(String field) { + this.field = field; + return this; + } + + public Builder addValue(String value) { + valueList.add(value); + return this; + } + + public Filter build() { + return new Filter(type, field, valueList); + } + } + + public String getType() { + return type; + } + + public String getField() { + return field; + } + + public List getValueList() { + return valueList; + } +} diff --git a/metatron/src/main/java/org/apache/zeppelin/metatron/message/Limits.java b/metatron/src/main/java/org/apache/zeppelin/metatron/message/Limits.java new file mode 100644 index 00000000000..d646a1be45e --- /dev/null +++ b/metatron/src/main/java/org/apache/zeppelin/metatron/message/Limits.java @@ -0,0 +1,31 @@ +package org.apache.zeppelin.metatron.message; + +import java.util.LinkedList; +import java.util.List; + +public class Limits { + List sort = new LinkedList<>(); + long limit; + + public Limits(long limit, List sort) { + this.limit = limit; + this.sort = sort; + } + + + public List getSort() { + return sort; + } + + public void setSort(List sort) { + this.sort = sort; + } + + public Limits(long limit) { + this.limit = limit; + } + + public long getLimit() { + return limit; + } +} diff --git a/metatron/src/main/java/org/apache/zeppelin/metatron/message/Link.java b/metatron/src/main/java/org/apache/zeppelin/metatron/message/Link.java new file mode 100644 index 00000000000..4a71d9bd42b --- /dev/null +++ b/metatron/src/main/java/org/apache/zeppelin/metatron/message/Link.java @@ -0,0 +1,14 @@ +package org.apache.zeppelin.metatron.message; + +public class Link { + String href; + boolean templated; + + public String getHref() { + return href; + } + + public boolean isTemplated() { + return templated; + } +} diff --git a/metatron/src/main/java/org/apache/zeppelin/metatron/message/Projection.java b/metatron/src/main/java/org/apache/zeppelin/metatron/message/Projection.java new file mode 100644 index 00000000000..9fa04ba0043 --- /dev/null +++ b/metatron/src/main/java/org/apache/zeppelin/metatron/message/Projection.java @@ -0,0 +1,19 @@ +package org.apache.zeppelin.metatron.message; + +public class Projection { + String type; + String name; + + public Projection(String type, String name) { + this.type = type; + this.name = name; + } + + public String getType() { + return type; + } + + public String getName() { + return name; + } +} diff --git a/metatron/src/main/java/org/apache/zeppelin/metatron/message/Record.java b/metatron/src/main/java/org/apache/zeppelin/metatron/message/Record.java new file mode 100644 index 00000000000..f2e6401e42b --- /dev/null +++ b/metatron/src/main/java/org/apache/zeppelin/metatron/message/Record.java @@ -0,0 +1,7 @@ +package org.apache.zeppelin.metatron.message; + +import java.util.HashMap; +import java.util.LinkedHashMap; + +public class Record extends LinkedHashMap { +} diff --git a/metatron/src/main/java/org/apache/zeppelin/metatron/message/SQLQuery.java b/metatron/src/main/java/org/apache/zeppelin/metatron/message/SQLQuery.java new file mode 100644 index 00000000000..5de1ddd72c4 --- /dev/null +++ b/metatron/src/main/java/org/apache/zeppelin/metatron/message/SQLQuery.java @@ -0,0 +1,69 @@ +package org.apache.zeppelin.metatron.message; + +import java.util.List; + +public class SQLQuery { + + static class Connection { + List datasources; + String implementor; + String authenticationType; + String username; + String password; + String hostname; + String port; + } + + String type; + Connection connection; + String database; + String query; + + public SQLQuery(String query) { + +// TODO: need to change metatron api for sql query with default connection info + + type = "QUERY"; + connection = new Connection(); + connection.implementor = "DRUID"; + connection.authenticationType = "MANUAL"; + connection.username = ""; + connection.password = ""; + connection.hostname = "localhost"; + connection.port = "8082"; + database = "druid"; + this.query = query; + } + + public String getType() { + return type; + } + + public void setType(String type) { + this.type = type; + } + + public Connection getConnection() { + return connection; + } + + public void setConnection(Connection connection) { + this.connection = connection; + } + + public String getDatabase() { + return database; + } + + public void setDatabase(String database) { + this.database = database; + } + + public String getQuery() { + return query; + } + + public void setQuery(String query) { + this.query = query; + } +} diff --git a/metatron/src/main/java/org/apache/zeppelin/metatron/message/SQLQueryResponse.java b/metatron/src/main/java/org/apache/zeppelin/metatron/message/SQLQueryResponse.java new file mode 100644 index 00000000000..9029f6cbd4a --- /dev/null +++ b/metatron/src/main/java/org/apache/zeppelin/metatron/message/SQLQueryResponse.java @@ -0,0 +1,25 @@ +package org.apache.zeppelin.metatron.message; + +import java.util.List; + +public class SQLQueryResponse { + + List fields; + List data; + + public List getFields() { + return fields; + } + + public void setFields(List fields) { + this.fields = fields; + } + + public List getData() { + return data; + } + + public void setData(List data) { + this.data = data; + } +} diff --git a/metatron/src/main/java/org/apache/zeppelin/metatron/message/User.java b/metatron/src/main/java/org/apache/zeppelin/metatron/message/User.java new file mode 100644 index 00000000000..5b7d035d3f1 --- /dev/null +++ b/metatron/src/main/java/org/apache/zeppelin/metatron/message/User.java @@ -0,0 +1,24 @@ +package org.apache.zeppelin.metatron.message; + +public class User { + String type; + String username; + String fullName; + String email; + + public String getType() { + return type; + } + + public String getUsername() { + return username; + } + + public String getFullName() { + return fullName; + } + + public String getEmail() { + return email; + } +} diff --git a/metatron/src/main/resources/interpreter-setting.json b/metatron/src/main/resources/interpreter-setting.json new file mode 100644 index 00000000000..25d028d7a52 --- /dev/null +++ b/metatron/src/main/resources/interpreter-setting.json @@ -0,0 +1,20 @@ +[ + { + "group": "metatron", + "name": "metatron", + "className": "org.apache.zeppelin.metatron.MetatronInterpreter", + "properties": { + "metatron.url": { + "envName": "METATRON_URL", + "propertyName": "metatron.url", + "defaultValue": "http://localhost:8180", + "description": "The url for Metatron", + "type": "string" + } + }, + "editor": { + "editOnDblClick": false, + "completionSupport": true + } + } +] diff --git a/metatron/src/test/java/org/apache/zeppelin/metatron/MetatronInterpreterTest.java b/metatron/src/test/java/org/apache/zeppelin/metatron/MetatronInterpreterTest.java new file mode 100644 index 00000000000..c3ef019133a --- /dev/null +++ b/metatron/src/test/java/org/apache/zeppelin/metatron/MetatronInterpreterTest.java @@ -0,0 +1,186 @@ +package org.apache.zeppelin.metatron; + +import org.antlr.v4.runtime.ANTLRErrorListener; +import org.apache.zeppelin.metatron.antlr.MetatronParser; +import org.junit.Test; +import org.junit.experimental.theories.DataPoint; +import org.junit.After; +import org.junit.Before; + +import org.apache.zeppelin.display.AngularObjectRegistry; +import org.apache.zeppelin.interpreter.InterpreterContext; +import org.apache.zeppelin.interpreter.InterpreterException; +import org.apache.zeppelin.interpreter.InterpreterResult; + +import java.io.IOException; +import java.util.List; +import java.util.Properties; + +import static org.junit.Assert.assertEquals; + +public class MetatronInterpreterTest { + +// +// @Test +// public void testParsenewStatement() throws IOException { +// // given +// MetatronInterpreter interpreter = new MetatronInterpreter(new Properties()); +// +// // when +//// MetatronParser parser = interpreter.parseMetatronExpr("source=sales Category=Furniture"); +// MetatronParser parser = interpreter.parseMetatronExpr("source=sales Category=Furniture | group sum(sales) by Category | rename sum(sales) as sum_sales | field Category, sum(sales)"); +//// MetatronParser parser = interpreter.parseMetatronExpr("abck adf;adf"); +// +// List antlrErrorListeners = (List) parser.getErrorListeners(); +// // then +// MetatronParser.ExprsContext exprs = parser.exprs(); +// +// MetatronParser.StmtContext stmtContext = exprs.stmt(); +// +// MetatronParser.BaseExpressionContext baseExpressionContext = stmtContext.baseExpression(); +// +// MetatronParser.SelectExpressionContext selectExpressionContext = baseExpressionContext.selectExpression(); +// +// String errorstring = selectExpressionContext.exception.toString(); +// +// String sample = exprs.getText(); +//// String sample2 = exprs.IDENTIFIER().getText(); +//// List stmts = parser.exprs().stmt(); +//// assertEquals(1, stmts.size()); +//// assertEquals("datasources", stmts.get(0).RESOURCE().getText()); +// } + +// @Test +// public void testParseSingleStatement() throws IOException { +// // given +// MetatronInterpreter interpreter = new MetatronInterpreter(new Properties()); +// +// // when +// MetatronParser parser = interpreter.parseMetatronExpr("describe datasources"); +// +// // then +// List stmts = parser.exprs().stmt(); +// assertEquals(1, stmts.size()); +//// assertEquals("datasources", stmts.get(0).RESOURCE().getText()); +// } +// +// @Test +// public void testParseMultipleStatements() throws IOException { +// // given +// MetatronInterpreter interpreter = new MetatronInterpreter(new Properties()); +// +// // when +// MetatronParser parser = interpreter.parseMetatronExpr("describe datasources;describe sales;"); +// +// // then +// List stmts = parser.exprs().stmt(); +// assertEquals(2, stmts.size()); +//// assertEquals("datasources", stmts.get(0).RESOURCE().getText()); +//// assertEquals("sales", stmts.get(1).RESOURCE().getText()); +// } +// +// @Test +// public void testInvalidGrammar() throws IOException { +// // given +// MetatronInterpreter interpreter = new MetatronInterpreter(new Properties()); +// +// // when +// MetatronParser parser = interpreter.parseMetatronExpr("something not valid"); +// +// // then +// List stmts = parser.exprs().stmt(); +// assertEquals(0, stmts.size()); +// } + + @DataPoint + private MetatronInterpreter interpreter; + + private static final String METATRON_TEST_URL = "http://localhost:8180"; + String accessToken = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE1NTgzNTc5NjAsInVzZXJfbmFtZSI6ImFkbWluIiwiYXV0aG9yaXRpZXMiOlsiUEVSTV9TWVNURU1fTUFOQUdFX0RBVEFTT1VSQ0UiLCJQRVJNX1NZU1RFTV9NQU5BR0VfUFJJVkFURV9XT1JLU1BBQ0UiLCJQRVJNX1NZU1RFTV9NQU5BR0VfVVNFUiIsIlBFUk1fU1lTVEVNX01BTkFHRV9TWVNURU0iLCJfX1BFUk1JU1NJT05fTUFOQUdFUiIsIl9fQURNSU4iLCJQRVJNX1NZU1RFTV9NQU5BR0VfU0hBUkVEX1dPUktTUEFDRSIsIl9fU0hBUkVEX1VTRVIiLCJQRVJNX1NZU1RFTV9WSUVXX1dPUktTUEFDRSIsIl9fREFUQV9NQU5BR0VSIiwiUEVSTV9TWVNURU1fTUFOQUdFX01FVEFEQVRBIiwiUEVSTV9TWVNURU1fTUFOQUdFX1dPUktTUEFDRSIsIl9fUFJJVkFURV9VU0VSIl0sImp0aSI6ImQ1OTliYmI0LTE3NGEtNDFiYy04YjA5LTdkMWE1M2JkYTM0YSIsImNsaWVudF9pZCI6InBvbGFyaXNfY2xpZW50Iiwic2NvcGUiOlsid3JpdGUiXX0.ncX7HjWoN7V8xExMqlmIlAR9ceQ9GSiROkSdZZyN1to"; + + @Before + public void setUp() throws Exception { + + final Properties props = new Properties(); + props.put(MetatronInterpreter.METATRON_URL, METATRON_TEST_URL); + interpreter = new MetatronInterpreter(props); + interpreter.open(); + + interpreter.getClient().setAccessToken(accessToken); + } + + @After + public void tearDown() throws InterpreterException { + interpreter.close(); + } + + + @Test + public void testShowDatasources() { + + final InterpreterContext ctx = buildContext("testShowDatasources"); + +// InterpreterResult res = interpreter.interpret("show datasources", ctx); + InterpreterResult res = interpreter.interpret("show datasources", null); + + assertEquals(InterpreterResult.Code.SUCCESS, res.code()); + String abc = res.message().get(0).getData(); + assertEquals("id\tname\ttype\tengine\tdescription\nds-gis-37\tsales\tsales_geo\tSales data (2011~2014)", res.message().get(0).getData()); + + } + + @Test + public void testShowDatasourceDetail() { + + final InterpreterContext ctx = buildContext("testShowDatasourceDetail"); + +// InterpreterResult res = interpreter.interpret("show datasources", ctx); + InterpreterResult res = interpreter.interpret("show sales", null); + + assertEquals(InterpreterResult.Code.SUCCESS, res.code()); + String abc = res.message().get(0).getData(); + assertEquals("id\tname\ttype\tengine\tdescription\nds-gis-37\tsales\tsales_geo\tSales data (2011~2014)", res.message().get(0).getData()); + + } + + @Test + public void testGetData() { + + final InterpreterContext ctx = buildContext("testGetData"); + + InterpreterResult res = interpreter.interpret("datasource=sales Category=Furniture limit=10 Category Sales", ctx); + + assertEquals(InterpreterResult.Code.SUCCESS, res.code()); + + } + + @Test + public void testSQLQuery() { + + final InterpreterContext ctx = buildContext("testSQLQuery"); + + InterpreterResult res = interpreter.interpret("select * from druid.sales", null); + + assertEquals(InterpreterResult.Code.SUCCESS, res.code()); + + } + + @Test + public void testComment() { + + final InterpreterContext ctx = buildContext("testComment"); + + InterpreterResult res = interpreter.interpret(" #this is test\nshow datasources", null); + + assertEquals(InterpreterResult.Code.SUCCESS, res.code()); + + } + + private InterpreterContext buildContext(String noteAndParagraphId) { + return InterpreterContext.builder() + .setNoteId(noteAndParagraphId) + .setParagraphId(noteAndParagraphId) + .setAngularObjectRegistry(new AngularObjectRegistry("metatron", null)) + .build(); + } +} diff --git a/pom.xml b/pom.xml index a66900dde59..fced66700c9 100644 --- a/pom.xml +++ b/pom.xml @@ -78,6 +78,7 @@ alluxio scio neo4j + metatron sap scalding java diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventClient.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventClient.java index 287095d9b06..3d58f337aaf 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventClient.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventClient.java @@ -115,39 +115,20 @@ public synchronized Object invokeMethod( Object[] params) { LOGGER.debug("Request Invoke method {} of Resource {}", methodName, resourceId.getName()); - return null; - // InvokeResourceMethodEventMessage invokeMethod = new InvokeResourceMethodEventMessage( - // resourceId, - // methodName, - // paramTypes, - // params, - // null); - // - // synchronized (getInvokeResponse) { - // // wait for previous response consumed - // while (getInvokeResponse.containsKey(invokeMethod)) { - // try { - // getInvokeResponse.wait(); - // } catch (InterruptedException e) { - // LOGGER.warn(e.getMessage(), e); - // } - // } - // // send request - // sendEvent(new RemoteInterpreterEvent( - // RemoteInterpreterEventType.RESOURCE_INVOKE_METHOD, - // invokeMethod.toJson())); - // // wait for response - // while (!getInvokeResponse.containsKey(invokeMethod)) { - // try { - // getInvokeResponse.wait(); - // } catch (InterruptedException e) { - // LOGGER.warn(e.getMessage(), e); - // } - // } - // Object o = getInvokeResponse.remove(invokeMethod); - // getInvokeResponse.notifyAll(); - // return o; - // } + InvokeResourceMethodEventMessage invokeMethod = new InvokeResourceMethodEventMessage( + resourceId, + methodName, + paramTypes, + params, + null); + try { + ByteBuffer buffer = intpEventServiceClient.invokeMethod(intpGroupId, invokeMethod.toJson()); + Object o = Resource.deserializeObject(buffer); + return o; + } catch (TException | IOException | ClassNotFoundException e) { + LOGGER.error("Failed to invoke method", e); + return null; + } } /** @@ -169,39 +150,24 @@ public synchronized Resource invokeMethod( String returnResourceName) { LOGGER.debug("Request Invoke method {} of Resource {}", methodName, resourceId.getName()); - return null; - // InvokeResourceMethodEventMessage invokeMethod = new InvokeResourceMethodEventMessage( - // resourceId, - // methodName, - // paramTypes, - // params, - // returnResourceName); - // - // synchronized (getInvokeResponse) { - // // wait for previous response consumed - // while (getInvokeResponse.containsKey(invokeMethod)) { - // try { - // getInvokeResponse.wait(); - // } catch (InterruptedException e) { - // LOGGER.warn(e.getMessage(), e); - // } - // } - // // send request - // sendEvent(new RemoteInterpreterEvent( - // RemoteInterpreterEventType.RESOURCE_INVOKE_METHOD, - // invokeMethod.toJson())); - // // wait for response - // while (!getInvokeResponse.containsKey(invokeMethod)) { - // try { - // getInvokeResponse.wait(); - // } catch (InterruptedException e) { - // LOGGER.warn(e.getMessage(), e); - // } - // } - // Resource o = (Resource) getInvokeResponse.remove(invokeMethod); - // getInvokeResponse.notifyAll(); - // return o; - // } + InvokeResourceMethodEventMessage invokeMethod = new InvokeResourceMethodEventMessage( + resourceId, + methodName, + paramTypes, + params, + returnResourceName); + + try { + ByteBuffer serializedResource = intpEventServiceClient.invokeMethod(intpGroupId, invokeMethod.toJson()); + Resource deserializedResource = (Resource) Resource.deserializeObject(serializedResource); + RemoteResource remoteResource = RemoteResource.fromJson(gson.toJson(deserializedResource)); + remoteResource.setResourcePoolConnector(this); + + return remoteResource; + } catch (TException | IOException | ClassNotFoundException e) { + LOGGER.error("Failed to invoke method", e); + return null; + } } public synchronized void onInterpreterOutputAppend( diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java index 037b2931861..30d87a2cab3 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java @@ -923,7 +923,6 @@ public List resourcePoolGetAll() throws TException { for (Resource r : resourceSet) { result.add(r.toJson()); } - return result; } @@ -957,7 +956,6 @@ public ByteBuffer resourceInvokeMethod( String noteId, String paragraphId, String resourceName, String invokeMessage) { InvokeResourceMethodEventMessage message = InvokeResourceMethodEventMessage.fromJson(invokeMessage); - Resource resource = resourcePool.get(noteId, paragraphId, resourceName, false); if (resource == null || resource.get() == null) { return ByteBuffer.allocate(0); @@ -971,13 +969,20 @@ public ByteBuffer resourceInvokeMethod( if (message.shouldPutResultIntoResourcePool()) { // if return resource name is specified, // then put result into resource pool - // and return empty byte buffer + // and return the Resource class instead of actual return object. resourcePool.put( noteId, paragraphId, message.returnResourceName, ret); - return ByteBuffer.allocate(0); + + Resource returnValResource = resourcePool.get(noteId, paragraphId, message.returnResourceName); + ByteBuffer serialized = Resource.serializeObject(returnValResource); + if (serialized == null) { + return ByteBuffer.allocate(0); + } else { + return serialized; + } } else { // if return resource name is not specified, // then return serialized result @@ -995,31 +1000,6 @@ public ByteBuffer resourceInvokeMethod( } } - // /** - // * Get payload of resource from remote - // * - // * @param invokeResourceMethodEventMessage json serialized InvokeResourcemethodEventMessage - // * @param object java serialized of the object - // * @throws TException - // */ - // @Override - // public void resourceResponseInvokeMethod( - // String invokeResourceMethodEventMessage, ByteBuffer object) throws TException { - // InvokeResourceMethodEventMessage message = - // InvokeResourceMethodEventMessage.fromJson(invokeResourceMethodEventMessage); - // - // if (message.shouldPutResultIntoResourcePool()) { - // Resource resource = resourcePool.get( - // message.resourceId.getNoteId(), - // message.resourceId.getParagraphId(), - // message.returnResourceName, - // true); - // eventClient.putResponseInvokeMethod(message, resource); - // } else { - // eventClient.putResponseInvokeMethod(message, object); - // } - // } - @Override public void angularRegistryPush(String registryAsString) throws TException { try { diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/util/ByteBufferUtils.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/util/ByteBufferUtils.java new file mode 100644 index 00000000000..aff758882fa --- /dev/null +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/util/ByteBufferUtils.java @@ -0,0 +1,37 @@ +/* + * 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. + */ +package org.apache.zeppelin.interpreter.util; + +import java.nio.ByteBuffer; +import java.nio.charset.Charset; + +public class ByteBufferUtils { + public static ByteBuffer stringToByteBuffer(String msg, Charset charset){ + return ByteBuffer.wrap(msg.getBytes(charset)); + } + + public static String ByteBufferToString(ByteBuffer buffer, Charset charset){ + byte[] bytes; + if(buffer.hasArray()) { + bytes = buffer.array(); + } else { + bytes = new byte[buffer.remaining()]; + buffer.get(bytes); + } + return new String(bytes, charset); + } +} diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/resource/RemoteResource.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/resource/RemoteResource.java index 874c1cbf8c5..19d84a03bb8 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/resource/RemoteResource.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/resource/RemoteResource.java @@ -17,15 +17,16 @@ package org.apache.zeppelin.resource; import com.google.gson.Gson; +import java.io.Serializable; import org.apache.zeppelin.common.JsonSerializable; /** * Resource that can retrieve data from remote */ -public class RemoteResource extends Resource implements JsonSerializable { +public class RemoteResource extends Resource implements JsonSerializable, Serializable { private static final Gson gson = new Gson(); - ResourcePoolConnector resourcePoolConnector; + transient ResourcePoolConnector resourcePoolConnector; RemoteResource(ResourceId resourceId, Object r) { super(null, resourceId, r); diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/resource/Resource.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/resource/Resource.java index ec95ffbfa44..c6717076b86 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/resource/Resource.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/resource/Resource.java @@ -17,6 +17,10 @@ package org.apache.zeppelin.resource; import com.google.gson.Gson; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.ParameterizedType; +import java.lang.reflect.Type; +import java.util.ArrayList; import org.apache.zeppelin.common.JsonSerializable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -33,7 +37,7 @@ /** * Information and reference to the resource */ -public class Resource implements JsonSerializable { +public class Resource implements JsonSerializable, Serializable { private static final Gson gson = new Gson(); private final transient Object r; @@ -111,6 +115,206 @@ public boolean isLocal() { return true; } + /** + * Invoke a method without param + * @param methodName + * @return + */ + public Object invokeMethod(String methodName) { + return invokeMethod(methodName, (Class []) null, (Object []) null); + } + + /** + * Invoke a method and store result in ResourcePool + * @param methodName + * @param returnResourceName + * @return + */ + public Resource invokeMethod(String methodName, String returnResourceName) { + return invokeMethod(methodName, (Class []) null, (Object []) null, returnResourceName); + } + + /** + * Invoke a method with automatic parameter type inference + * @param methodName + * @param params + * @return + * @throws ClassNotFoundException + */ + public Object invokeMethod(String methodName, Object [] params) + throws ClassNotFoundException { + return invokeMethod(methodName, (Type[]) null, params); + } + + /** + * Invoke a method with automatic parameter type inference + * @param methodName + * @param params python interpreter convert python array '[]' to ArrayList through py4j + * @return + * @throws ClassNotFoundException + */ + public Object invokeMethod( + String methodName, ArrayList params) + throws ClassNotFoundException { + Object[] paramsArray = params.toArray(new Object[]{}); + return invokeMethod(methodName, paramsArray); + } + + /** + * Invoke a method with automatic parameter type inference and store result in ResourcePool + * @param methodName + * @param params + * @param returnResourceName + * @return + * @throws ClassNotFoundException + */ + public Resource invokeMethod(String methodName, Object [] params, String returnResourceName) + throws ClassNotFoundException { + return (Resource) invokeMethod(methodName, (Type[]) null, params, returnResourceName); + } + + /** + * Invoke a method with automatic parameter type inference and store result in ResourcePool + * @param methodName + * @param params python interpreter convert python array '[]' to ArrayList through py4j + * @param returnResourceName + * @return + * @throws ClassNotFoundException + */ + public Resource invokeMethod( + String methodName, ArrayList params, String returnResourceName) + throws ClassNotFoundException { + Object[] paramsArray = params.toArray(new Object[]{}); + return invokeMethod(methodName, paramsArray, returnResourceName); + } + + /** + * Invoke a method with given parameter class names + * @param methodName + * @param paramTypes list of fully qualified class name + * @param params + * @return + * @throws ClassNotFoundException + */ + public Object invokeMethod( + String methodName, String[] paramTypes, Object[] params) + throws ClassNotFoundException { + Type [] types = typeFromName(paramTypes); + return invokeMethod(methodName, types, params); + } + + /** + * Invoke a method with given parameter class names + * @param methodName + * @param paramTypes list of fully qualified class name. python interpreter convert python array '[]' to ArrayList through py4j + * @param params python interpreter convert python array '[]' to ArrayList through py4j + * @return + * @throws ClassNotFoundException + */ + public Object invokeMethod( + String methodName, ArrayList paramTypes, ArrayList params) + throws ClassNotFoundException { + String[] paramTypesArray = paramTypes.toArray(new String[]{}); + Object[] paramsArray = params.toArray(new Object[]{}); + return invokeMethod(methodName, paramTypesArray, paramsArray); + } + + /** + * Invoke a method with given parameter class names and store result in ResourcePool + * @param methodName + * @param paramTypes + * @param params + * @param returnResourceName + * @return + * @throws ClassNotFoundException + */ + public Resource invokeMethod( + String methodName, String[] paramTypes, Object[] params, String returnResourceName) + throws ClassNotFoundException { + Type [] types = typeFromName(paramTypes); + return (Resource) invokeMethod(methodName, types, params, returnResourceName); + } + + + public Resource invokeMethod( + String methodName, ArrayList paramTypes, ArrayList params, String returnResourceName) + throws ClassNotFoundException { + String[] paramTypesArray = paramTypes.toArray(new String[]{}); + Object[] paramsArray = params.toArray(new Object[]{}); + return invokeMethod(methodName, paramTypesArray, paramsArray, returnResourceName); + } + + /** + * Invoke a method with give parameter types + * @param methodName + * @param types + * @param params + * @return + * @throws ClassNotFoundException + */ + public Object invokeMethod( + String methodName, Type[] types, Object[] params) + throws ClassNotFoundException { + return invokeMethod(methodName, types, params, null); + } + + /** + * Invoke a method with given parameter type and store result in ResourcePool + * @param methodName + * @param types + * @param params + * @param returnResourceName + * @return + * @throws ClassNotFoundException + */ + public Object invokeMethod( + String methodName, Type[] types, Object[] params, String returnResourceName) throws ClassNotFoundException { + Type[] methodTypes = null; + Object [] methodParams = null; + if (types != null) { + methodTypes = types; + methodParams = params; + } else { + // inference method param types + boolean found = false; + Method[] methods = r.getClass().getDeclaredMethods(); + for (Method m : methods) { + if (!m.getName().equals(methodName)) { + continue; + } + Type[] paramTypes = m.getGenericParameterTypes(); + Object[] paramValues = new Object[paramTypes.length]; + + int pidx = 0; + for (int i = 0; i < paramTypes.length; i++) { + if (pidx == params.length) { // not enough param for this method signature + continue; + } else { + paramValues[i] = params[pidx++]; + } + } + + if (pidx == params.length) { // param number does not match + found = true; + methodParams = paramValues; + methodTypes = paramTypes; + break; + } + } + + if (!found) { + throw new ClassNotFoundException("No method found for given parameters"); + } + } + + Class[] classes = classFromType(methodTypes); + + if (returnResourceName == null) { + return invokeMethod(methodName, classes, convertParams(methodTypes, methodParams)); + } else { + return invokeMethod(methodName, classes, convertParams(methodTypes, methodParams), returnResourceName); + } + } /** * Call a method of the object that this resource holds @@ -222,4 +426,111 @@ public String toJson() { public static Resource fromJson(String json) { return gson.fromJson(json, Resource.class); } + + private ParameterizedType [] typeFromName(String [] classNames) throws ClassNotFoundException { + if (classNames == null) { + return null; + } + ParameterizedType[] types = new ParameterizedType[classNames.length]; + for (int i = 0; i < classNames.length; i++) { + types[i] = typeFromName(classNames[i]); + } + return types; + } + + private ParameterizedType typeFromName(String commaSeparatedClasses) throws ClassNotFoundException { + String[] classNames = commaSeparatedClasses.split(","); + Class [] arguments; + + if (classNames.length > 1) { + arguments = new Class[classNames.length - 1]; + for (int i = 1; i < classNames.length; i++) { + arguments[i - 1] = loadClass(classNames[i]); + } + } else { + arguments = new Class[0]; + } + + Class rawType = loadClass(classNames[0]); + + return new ParameterizedType() { + @Override + public Type[] getActualTypeArguments() { + return arguments; + } + + @Override + public Type getRawType() { + return rawType; + } + + @Override + public Type getOwnerType() { + return null; + } + }; + } + + private Class [] classFromType(Type[] types) throws ClassNotFoundException { + Class[] cls = new Class[types.length]; + for (int i = 0; i < types.length; i++) { + if (types[i] instanceof ParameterizedType) { + String typeName = ((ParameterizedType) types[i]).getRawType().getTypeName(); + cls[i] = loadClass(typeName); + } else { + cls[i] = loadClass(types[i].getTypeName()); + } + } + return cls; + } + + + private Object [] convertParams(Type[] types, Object [] params) { + Object [] converted = new Object[types.length]; + + for (int i = 0; i < types.length; i++) { + Type type = types[i]; + String typeName; + if (type instanceof ParameterizedType) { + typeName = ((ParameterizedType) type).getRawType().getTypeName(); + } else { + typeName = type.getTypeName(); + } + + Object param = params[i]; + if (param == null) { + converted[i] = null; + } else if (param.getClass().getName().equals(typeName)) { + converted[i] = param; + } else { + // try to convert param + converted[i] = gson.fromJson(gson.toJson(param), type); + } + } + + return converted; + } + + private Class loadClass(String className) throws ClassNotFoundException { + switch(className) { + case "byte": + return byte.class; + case "short": + return short.class; + case "int": + return int.class; + case "long": + return long.class; + case "float": + return float.class; + case "double": + return double.class; + case "boolean": + return boolean.class; + case "char": + return char.class; + default: + return getClass().getClassLoader().loadClass(className); + } + } } diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/resource/ResourceId.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/resource/ResourceId.java index bef9e3fd306..ce06b73a839 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/resource/ResourceId.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/resource/ResourceId.java @@ -17,12 +17,13 @@ package org.apache.zeppelin.resource; import com.google.gson.Gson; +import java.io.Serializable; import org.apache.zeppelin.common.JsonSerializable; /** * Identifying resource */ -public class ResourceId implements JsonSerializable { +public class ResourceId implements JsonSerializable, Serializable { private static final Gson gson = new Gson(); private final String resourcePoolId; diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/tabledata/ProxyRowIterator.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/tabledata/ProxyRowIterator.java index 8a59098d8ee..ceb122c3cbe 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/tabledata/ProxyRowIterator.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/tabledata/ProxyRowIterator.java @@ -33,13 +33,13 @@ public ProxyRowIterator(Resource rows) { @Override public boolean hasNext() { - rows.invokeMethod("hasNext", null, null); + rows.invokeMethod("hasNext"); return false; } @Override public Row next() { - return (Row) rows.invokeMethod("next", null, null); + return (Row) rows.invokeMethod("next"); } @Override diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/tabledata/TableDataProxy.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/tabledata/TableDataProxy.java index 19265287ae5..bb1f8421755 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/tabledata/TableDataProxy.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/tabledata/TableDataProxy.java @@ -33,13 +33,13 @@ public TableDataProxy(Resource tableDataRemoteResource) { @Override public ColumnDef[] columns() { return (ColumnDef[]) resource.invokeMethod( - "columns", null, null); + "columns"); } @Override public Iterator rows() { String resourceName = resource.getResourceId().getName() + ".rows"; - Resource rows = resource.invokeMethod("rows", null, null, resourceName); + Resource rows = resource.invokeMethod("rows", resourceName); ProxyRowIterator it = new ProxyRowIterator(rows); return it; diff --git a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/util/ByteBufferUtilTest.java b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/util/ByteBufferUtilTest.java new file mode 100644 index 00000000000..bfd40b27f52 --- /dev/null +++ b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/util/ByteBufferUtilTest.java @@ -0,0 +1,17 @@ +package org.apache.zeppelin.interpreter.util; + +import java.nio.ByteBuffer; +import java.nio.charset.Charset; +import org.junit.Test; + +import static org.junit.Assert.assertEquals; + +public class ByteBufferUtilTest { + + @Test + public void fromByteBufferToByteBuffer() { + String str = "Hello world"; + ByteBuffer byteBuffer = ByteBufferUtils.stringToByteBuffer(str, Charset.defaultCharset()); + assertEquals(str, ByteBufferUtils.ByteBufferToString(byteBuffer, Charset.defaultCharset())); + } +} diff --git a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/resource/ResourceTest.java b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/resource/ResourceTest.java index fb8b27131f6..211d85dbf8b 100644 --- a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/resource/ResourceTest.java +++ b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/resource/ResourceTest.java @@ -16,6 +16,9 @@ */ package org.apache.zeppelin.resource; +import java.lang.reflect.InvocationTargetException; +import java.util.ArrayList; +import java.util.Arrays; import org.junit.Test; import java.io.IOException; @@ -32,4 +35,37 @@ public void testSerializeDeserialize() throws IOException, ClassNotFoundExceptio ByteBuffer buffer = Resource.serializeObject("hello"); assertEquals("hello", Resource.deserializeObject(buffer)); } + + @Test + public void testInvokeMethod_shouldAbleToInvokeMethodWithNoParams() { + Resource r = new Resource(null, new ResourceId("pool1", "name1"), "object"); + assertEquals(6, r.invokeMethod("length")); + assertEquals(6, r.invokeMethod("length", new Class[]{}, new Object[]{})); + } + + @Test + public void testInvokeMethod_shouldAbleToInvokeMethodWithTypeInference() throws ClassNotFoundException, NoSuchMethodException, InvocationTargetException, IllegalAccessException { + Resource r = new Resource(null, new ResourceId("pool1", "name1"), "object"); + assertEquals("ect", r.invokeMethod("substring", new Object[]{3})); + assertEquals(true, r.invokeMethod("startsWith", new Object[]{"obj"})); + + assertEquals("ect", r.invokeMethod("substring", new ArrayList<>(Arrays.asList(3)))); + assertEquals(true, r.invokeMethod("startsWith", new ArrayList<>(Arrays.asList("obj")))); + } + + @Test + public void testInvokeMethod_shouldAbleToInvokeMethodWithParamClassName() throws ClassNotFoundException, NoSuchMethodException, InvocationTargetException, IllegalAccessException { + Resource r = new Resource(null, new ResourceId("pool1", "name1"), "object"); + assertEquals("ect", r.invokeMethod("substring", new String[]{"int"}, new Object[]{3})); + assertEquals(true, r.invokeMethod("startsWith", new String[]{"java.lang.String"}, new Object[]{"obj"})); + + assertEquals("ect", r.invokeMethod("substring", new ArrayList<>(Arrays.asList("int")), new ArrayList<>(Arrays.asList(3)))); + assertEquals(true, r.invokeMethod("startsWith", new ArrayList<>(Arrays.asList("java.lang.String")), new ArrayList<>(Arrays.asList("obj")))); + } + + @Test + public void testInvokeMethod_shouldAbleToInvokeMethodWithClass() throws ClassNotFoundException, NoSuchMethodException, InvocationTargetException, IllegalAccessException { + Resource r = new Resource(null, new ResourceId("pool1", "name1"), "object"); + assertEquals(true, r.invokeMethod("startsWith", new Class[]{ java.lang.String.class }, new Object[]{"obj"})); + } } diff --git a/zeppelin-server/pom.xml b/zeppelin-server/pom.xml index d30060ddc46..50b1665ede7 100644 --- a/zeppelin-server/pom.xml +++ b/zeppelin-server/pom.xml @@ -268,6 +268,36 @@ ${scala.version} + + io.buji + buji-pac4j + 4.0.0 + + + + org.pac4j + pac4j-oauth + 3.0.2 + + + com.fasterxml.jackson.core + jackson-databind + + + + + + org.pac4j + pac4j-http + 3.0.2 + + + com.fasterxml.jackson.core + jackson-databind + + + + junit diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/realm/MetatronRealm.java b/zeppelin-server/src/main/java/org/apache/zeppelin/realm/MetatronRealm.java new file mode 100644 index 00000000000..7307fbe2d98 --- /dev/null +++ b/zeppelin-server/src/main/java/org/apache/zeppelin/realm/MetatronRealm.java @@ -0,0 +1,122 @@ +package org.apache.zeppelin.realm; + +import com.google.common.collect.ImmutableList; +import com.google.gson.Gson; +import java.io.IOException; +import org.apache.commons.io.Charsets; +import org.apache.commons.io.IOUtils; +import org.apache.http.HttpResponse; +import org.apache.http.client.entity.UrlEncodedFormEntity; +import org.apache.http.client.methods.CloseableHttpResponse; +import org.apache.http.client.methods.HttpGet; +import org.apache.http.client.methods.HttpPost; +import org.apache.http.client.methods.HttpUriRequest; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.client.HttpClients; +import org.apache.http.message.BasicNameValuePair; +import org.apache.shiro.authc.AuthenticationException; +import org.apache.shiro.authc.AuthenticationInfo; +import org.apache.shiro.authc.AuthenticationToken; +import org.apache.shiro.authc.UsernamePasswordToken; +import org.apache.shiro.authz.AuthorizationInfo; +import org.apache.shiro.authc.SimpleAuthenticationInfo; +import org.apache.shiro.realm.AuthorizingRealm; +import org.apache.shiro.subject.PrincipalCollection; +import org.apache.zeppelin.server.ZeppelinServer; +import org.apache.zeppelin.user.UserCredentials; +import org.apache.zeppelin.user.UsernamePassword; +import org.eclipse.jetty.util.security.Credential; + +import javax.inject.Inject; + +public class MetatronRealm extends AuthorizingRealm { + String authUrl; // http(s)://localhost:8180/oauth/token + Gson gson = new Gson(); + CloseableHttpClient httpClient = HttpClients.createDefault(); + + public MetatronRealm() { + super(); + } + + @Override + protected AuthorizationInfo doGetAuthorizationInfo(PrincipalCollection principalCollection) { + return null; + } + + @Override + protected AuthenticationInfo doGetAuthenticationInfo(AuthenticationToken authenticationToken) throws AuthenticationException { + UsernamePasswordToken token = (UsernamePasswordToken) authenticationToken; + AuthResponse authResp; + try { + authResp = auth(token.getUsername(), new StringBuilder().append(token.getPassword()).toString()); + } catch (IOException e) { + throw new AuthenticationException(e); + } + + if (!authResp.isAuthenticated()) { + throw new AuthenticationException("Invalid id or password"); + } + + return new SimpleAuthenticationInfo(token.getUsername(), token.getPassword(), MetatronRealm.class.getName()); + } + + public class AuthResponse { + String access_token; + String token_type; + long expires_in; + String scope; + String jti; + + public boolean isAuthenticated() { + return access_token != null; + } + + public String getAccess_token() { + return access_token; + } + + public String getToken_type() { + return token_type; + } + + public long getExpires_in() { + return expires_in; + } + + public String getScope() { + return scope; + } + + public String getJti() { + return jti; + } + } + + AuthResponse auth(String username, String password) throws IOException { + HttpPost post = new HttpPost(authUrl); + post.setHeader("Content-Type", "application/x-www-form-urlencoded"); + post.setHeader("Authorization", "Basic cG9sYXJpc19jbGllbnQ6cG9sYXJpcw=="); + + post.setEntity(new UrlEncodedFormEntity(ImmutableList.of( + new BasicNameValuePair("grant_type", "password"), + new BasicNameValuePair("scope", "write"), + new BasicNameValuePair("username", username), + new BasicNameValuePair("password", password) + ))); + CloseableHttpResponse resp = httpClient.execute(post); + + String result = IOUtils.toString(resp.getEntity().getContent(), Charsets.UTF_8); + AuthResponse authResp = gson.fromJson(result, AuthResponse.class); + + // put user credential, so interpreter can read this information + UserCredentials userCredentials = new UserCredentials(); + userCredentials.putUsernamePassword("token", new UsernamePassword(username, authResp.access_token)); + ZeppelinServer.notebook.getCredentials().putUserCredentials(username, userCredentials); + + return authResp; + } + + public void setAuthUrl(String authUrl) { + this.authUrl = authUrl; + } +} diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java index bd612d6e842..8970ad8d3df 100644 --- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java +++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java @@ -325,6 +325,13 @@ public ByteBuffer getResource(String resourceIdJson) throws TException { return obj; } + /** + * + * @param intpGroupId caller interpreter group id + * @param invokeMethodJson invoke information + * @return + * @throws TException + */ @Override public ByteBuffer invokeMethod(String intpGroupId, String invokeMethodJson) throws TException { InvokeResourceMethodEventMessage invokeMethodMessage = @@ -337,7 +344,7 @@ public ByteBuffer invokeMethod(String intpGroupId, String invokeMethodJson) thro try { obj = Resource.serializeObject(ret); } catch (IOException e) { - e.printStackTrace(); + LOGGER.error("invokeMethod failed", e); } } return obj; @@ -377,10 +384,8 @@ private Object invokeResourceMethod(String intpGroupId, LOGGER.error("no resource pool"); return null; } - } else if (interpreterSettingManager.getInterpreterGroupById(intpGroupId) - .getInterpreterProcess().isRunning()) { - ByteBuffer res = interpreterSettingManager.getInterpreterGroupById(intpGroupId) - .getInterpreterProcess().callRemoteFunction( + } else if (remoteInterpreterProcess.isRunning()) { + ByteBuffer res = remoteInterpreterProcess.callRemoteFunction( new RemoteInterpreterProcess.RemoteFunction() { @Override public ByteBuffer call(RemoteInterpreterService.Client client) throws Exception { diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Notebook.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Notebook.java index 671a0cb4b11..23bdf777871 100644 --- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Notebook.java +++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Notebook.java @@ -713,4 +713,8 @@ public Boolean isRevisionSupported() { return false; } } + + public Credentials getCredentials() { + return credentials; + } } diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterResourcePool.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterResourcePool.java index d9000318dd7..c01bbd2c608 100644 --- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterResourcePool.java +++ b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/mock/MockInterpreterResourcePool.java @@ -92,11 +92,11 @@ public InterpreterResult interpret(String st, InterpreterContext context) { Resource resource = resourcePool.get(noteId, paragraphId, name); LOGGER.info("Resource: " + resource); if (stmt.length >=4) { - Resource res = resource.invokeMethod(value, null, null, stmt[3]); + Resource res = resource.invokeMethod(value, stmt[3]); LOGGER.info("After invokeMethod: " + resource); ret = res.get(); } else { - ret = resource.invokeMethod(value, null, null); + ret = resource.invokeMethod(value); LOGGER.info("After invokeMethod: " + ret); } } diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/resource/DistributedResourcePoolTest.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/resource/DistributedResourcePoolTest.java index 925515e60f5..54c09b6fa89 100644 --- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/resource/DistributedResourcePoolTest.java +++ b/zeppelin-zengine/src/test/java/org/apache/zeppelin/resource/DistributedResourcePoolTest.java @@ -20,6 +20,7 @@ import org.apache.zeppelin.interpreter.AbstractInterpreterTest; import org.apache.zeppelin.interpreter.InterpreterContext; import org.apache.zeppelin.interpreter.InterpreterException; +import org.apache.zeppelin.interpreter.InterpreterOption; import org.apache.zeppelin.interpreter.InterpreterResult; import org.apache.zeppelin.interpreter.InterpreterSetting; import org.apache.zeppelin.interpreter.remote.RemoteInterpreter; @@ -27,6 +28,7 @@ import org.junit.Before; import org.junit.Test; +import static org.apache.zeppelin.interpreter.InterpreterOption.ISOLATED; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; @@ -44,8 +46,9 @@ public class DistributedResourcePoolTest extends AbstractInterpreterTest { public void setUp() throws Exception { super.setUp(); InterpreterSetting interpreterSetting = interpreterSettingManager.getByName("mock_resource_pool"); + interpreterSetting.getOption().setPerNote(ISOLATED); intp1 = (RemoteInterpreter) interpreterSetting.getInterpreter("user1", "note1", "mock_resource_pool"); - intp2 = (RemoteInterpreter) interpreterSetting.getInterpreter("user2", "note1", "mock_resource_pool"); + intp2 = (RemoteInterpreter) interpreterSetting.getInterpreter("user2", "note2", "mock_resource_pool"); context = InterpreterContext.builder() .setNoteId("note")