| /** |
| * 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.falcon.client; |
| |
| import com.sun.jersey.api.client.Client; |
| import com.sun.jersey.api.client.ClientResponse; |
| import com.sun.jersey.api.client.WebResource; |
| import com.sun.jersey.api.client.config.DefaultClientConfig; |
| import com.sun.jersey.client.urlconnection.HTTPSProperties; |
| import org.apache.commons.io.IOUtils; |
| import org.apache.commons.lang.StringUtils; |
| import org.apache.commons.net.util.TrustManagerUtils; |
| import org.apache.falcon.LifeCycle; |
| import org.apache.falcon.entity.v0.EntityType; |
| import org.apache.falcon.entity.v0.SchemaHelper; |
| import org.apache.falcon.resource.APIResult; |
| import org.apache.falcon.resource.EntityList; |
| import org.apache.falcon.resource.EntitySummaryResult; |
| import org.apache.falcon.resource.InstancesResult; |
| import org.apache.falcon.resource.InstancesSummaryResult; |
| import org.apache.hadoop.security.authentication.client.AuthenticatedURL; |
| import org.apache.hadoop.security.authentication.client.KerberosAuthenticator; |
| import org.apache.hadoop.security.authentication.client.PseudoAuthenticator; |
| |
| import javax.net.ssl.HostnameVerifier; |
| import javax.net.ssl.HttpsURLConnection; |
| import javax.net.ssl.SSLContext; |
| import javax.net.ssl.SSLSession; |
| import javax.net.ssl.TrustManager; |
| import javax.ws.rs.HttpMethod; |
| import javax.ws.rs.core.MediaType; |
| import javax.ws.rs.core.Response; |
| import javax.ws.rs.core.UriBuilder; |
| import java.io.BufferedReader; |
| import java.io.ByteArrayInputStream; |
| import java.io.FileInputStream; |
| import java.io.FileNotFoundException; |
| import java.io.FileReader; |
| import java.io.IOException; |
| import java.io.InputStream; |
| import java.io.UnsupportedEncodingException; |
| import java.net.URL; |
| import java.security.SecureRandom; |
| import java.util.Date; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Properties; |
| |
| /** |
| * Client API to submit and manage Falcon Entities (Cluster, Feed, Process) jobs |
| * against an Falcon instance. |
| */ |
| public class FalconClient { |
| |
| public static final String WS_HEADER_PREFIX = "header:"; |
| public static final String USER = System.getProperty("user.name"); |
| public static final String AUTH_URL = "api/options?" + PseudoAuthenticator.USER_NAME + "=" + USER; |
| |
| private static final String FALCON_INSTANCE_ACTION_CLUSTERS = "falcon.instance.action.clusters"; |
| private static final String FALCON_INSTANCE_SOURCE_CLUSTERS = "falcon.instance.source.clusters"; |
| |
| /** |
| * Name of the HTTP cookie used for the authentication token between the client and the server. |
| */ |
| public static final String AUTH_COOKIE = "hadoop.auth"; |
| private static final String AUTH_COOKIE_EQ = AUTH_COOKIE + "="; |
| private static final KerberosAuthenticator AUTHENTICATOR = new KerberosAuthenticator(); |
| |
| public static final int DEFAULT_NUM_RESULTS = 10; |
| |
| public static final HostnameVerifier ALL_TRUSTING_HOSTNAME_VERIFIER = new HostnameVerifier() { |
| @Override |
| public boolean verify(String hostname, SSLSession sslSession) { |
| return true; |
| } |
| }; |
| |
| private final WebResource service; |
| private final AuthenticatedURL.Token authenticationToken; |
| |
| private final Properties clientProperties; |
| |
| /** |
| * Create a Falcon client instance. |
| * |
| * @param falconUrl of the server to which client interacts |
| * @throws FalconCLIException - If unable to initialize SSL Props |
| */ |
| public FalconClient(String falconUrl) throws FalconCLIException { |
| this(falconUrl, new Properties()); |
| } |
| |
| /** |
| * Create a Falcon client instance. |
| * |
| * @param falconUrl of the server to which client interacts |
| * @param properties client properties |
| * @throws FalconCLIException - If unable to initialize SSL Props |
| */ |
| public FalconClient(String falconUrl, Properties properties) throws FalconCLIException { |
| try { |
| String baseUrl = notEmpty(falconUrl, "FalconUrl"); |
| if (!baseUrl.endsWith("/")) { |
| baseUrl += "/"; |
| } |
| this.clientProperties = properties; |
| SSLContext sslContext = getSslContext(); |
| DefaultClientConfig config = new DefaultClientConfig(); |
| config.getProperties().put(HTTPSProperties.PROPERTY_HTTPS_PROPERTIES, |
| new HTTPSProperties(ALL_TRUSTING_HOSTNAME_VERIFIER, sslContext) |
| ); |
| Client client = Client.create(config); |
| client.setConnectTimeout(Integer.parseInt(clientProperties.getProperty("falcon.connect.timeout", |
| "180000"))); |
| client.setReadTimeout(Integer.parseInt(clientProperties.getProperty("falcon.read.timeout", "180000"))); |
| service = client.resource(UriBuilder.fromUri(baseUrl).build()); |
| client.resource(UriBuilder.fromUri(baseUrl).build()); |
| authenticationToken = getToken(baseUrl); |
| } catch (Exception e) { |
| throw new FalconCLIException("Unable to initialize Falcon Client object", e); |
| } |
| } |
| |
| private static SSLContext getSslContext() throws Exception { |
| SSLContext sslContext = SSLContext.getInstance("SSL"); |
| sslContext.init( |
| null, |
| new TrustManager[]{TrustManagerUtils.getValidateServerCertificateTrustManager()}, |
| new SecureRandom()); |
| return sslContext; |
| } |
| |
| public Properties getClientProperties() { |
| return clientProperties; |
| } |
| |
| public static AuthenticatedURL.Token getToken(String baseUrl) throws FalconCLIException { |
| AuthenticatedURL.Token currentToken = new AuthenticatedURL.Token(); |
| try { |
| URL url = new URL(baseUrl + AUTH_URL); |
| // using KerberosAuthenticator which falls back to PsuedoAuthenticator |
| // instead of passing authentication type from the command line - bad factory |
| HttpsURLConnection.setDefaultSSLSocketFactory(getSslContext().getSocketFactory()); |
| HttpsURLConnection.setDefaultHostnameVerifier(ALL_TRUSTING_HOSTNAME_VERIFIER); |
| new AuthenticatedURL(AUTHENTICATOR).openConnection(url, currentToken); |
| } catch (Exception ex) { |
| throw new FalconCLIException("Could not authenticate, " + ex.getMessage(), ex); |
| } |
| |
| return currentToken; |
| } |
| |
| /** |
| * Methods allowed on Entity Resources. |
| */ |
| protected static enum Entities { |
| VALIDATE("api/entities/validate/", HttpMethod.POST, MediaType.TEXT_XML), |
| SUBMIT("api/entities/submit/", HttpMethod.POST, MediaType.TEXT_XML), |
| UPDATE("api/entities/update/", HttpMethod.POST, MediaType.TEXT_XML), |
| SUBMITandSCHEDULE("api/entities/submitAndSchedule/", HttpMethod.POST, MediaType.TEXT_XML), |
| SCHEDULE("api/entities/schedule/", HttpMethod.POST, MediaType.TEXT_XML), |
| SUSPEND("api/entities/suspend/", HttpMethod.POST, MediaType.TEXT_XML), |
| RESUME("api/entities/resume/", HttpMethod.POST, MediaType.TEXT_XML), |
| DELETE("api/entities/delete/", HttpMethod.DELETE, MediaType.TEXT_XML), |
| STATUS("api/entities/status/", HttpMethod.GET, MediaType.TEXT_XML), |
| DEFINITION("api/entities/definition/", HttpMethod.GET, MediaType.TEXT_XML), |
| LIST("api/entities/list/", HttpMethod.GET, MediaType.TEXT_XML), |
| SUMMARY("api/entities/summary", HttpMethod.GET, MediaType.APPLICATION_JSON), |
| DEPENDENCY("api/entities/dependencies/", HttpMethod.GET, MediaType.TEXT_XML); |
| |
| private String path; |
| private String method; |
| private String mimeType; |
| |
| Entities(String path, String method, String mimeType) { |
| this.path = path; |
| this.method = method; |
| this.mimeType = mimeType; |
| } |
| } |
| |
| /** |
| * Methods allowed on Process Instance Resources. |
| */ |
| protected static enum Instances { |
| RUNNING("api/instance/running/", HttpMethod.GET, MediaType.APPLICATION_JSON), |
| STATUS("api/instance/status/", HttpMethod.GET, MediaType.APPLICATION_JSON), |
| LIST("api/instance/list", HttpMethod.GET, MediaType.APPLICATION_JSON), |
| KILL("api/instance/kill/", HttpMethod.POST, MediaType.APPLICATION_JSON), |
| SUSPEND("api/instance/suspend/", HttpMethod.POST, MediaType.APPLICATION_JSON), |
| RESUME("api/instance/resume/", HttpMethod.POST, MediaType.APPLICATION_JSON), |
| RERUN("api/instance/rerun/", HttpMethod.POST, MediaType.APPLICATION_JSON), |
| LOG("api/instance/logs/", HttpMethod.GET, MediaType.APPLICATION_JSON), |
| SUMMARY("api/instance/summary/", HttpMethod.GET, MediaType.APPLICATION_JSON), |
| PARAMS("api/instance/params/", HttpMethod.GET, MediaType.APPLICATION_JSON); |
| |
| private String path; |
| private String method; |
| private String mimeType; |
| |
| Instances(String path, String method, String mimeType) { |
| this.path = path; |
| this.method = method; |
| this.mimeType = mimeType; |
| } |
| } |
| |
| protected static enum AdminOperations { |
| |
| STACK("api/admin/stack", HttpMethod.GET, MediaType.TEXT_PLAIN), |
| VERSION("api/admin/version", HttpMethod.GET, MediaType.APPLICATION_JSON); |
| |
| private String path; |
| private String method; |
| private String mimeType; |
| |
| AdminOperations(String path, String method, String mimeType) { |
| this.path = path; |
| this.method = method; |
| this.mimeType = mimeType; |
| } |
| } |
| |
| public String notEmpty(String str, String name) { |
| if (str == null) { |
| |
| throw new IllegalArgumentException(name + " cannot be null"); |
| } |
| if (str.length() == 0) { |
| throw new IllegalArgumentException(name + " cannot be empty"); |
| } |
| return str; |
| } |
| |
| public String schedule(String entityType, String entityName, String colo) |
| throws FalconCLIException { |
| |
| return sendEntityRequest(Entities.SCHEDULE, entityType, entityName, |
| colo); |
| |
| } |
| |
| public String suspend(String entityType, String entityName, String colo) |
| throws FalconCLIException { |
| |
| return sendEntityRequest(Entities.SUSPEND, entityType, entityName, colo); |
| |
| } |
| |
| public String resume(String entityType, String entityName, String colo) |
| throws FalconCLIException { |
| |
| return sendEntityRequest(Entities.RESUME, entityType, entityName, colo); |
| |
| } |
| |
| public String delete(String entityType, String entityName) |
| throws FalconCLIException { |
| |
| return sendEntityRequest(Entities.DELETE, entityType, entityName, null); |
| |
| } |
| |
| public String validate(String entityType, String filePath) |
| throws FalconCLIException { |
| |
| InputStream entityStream = getServletInputStream(filePath); |
| return sendEntityRequestWithObject(Entities.VALIDATE, entityType, |
| entityStream, null); |
| } |
| |
| public String submit(String entityType, String filePath) |
| throws FalconCLIException { |
| |
| InputStream entityStream = getServletInputStream(filePath); |
| return sendEntityRequestWithObject(Entities.SUBMIT, entityType, |
| entityStream, null); |
| } |
| |
| public String update(String entityType, String entityName, String filePath, Date effectiveTime) |
| throws FalconCLIException { |
| InputStream entityStream = getServletInputStream(filePath); |
| Entities operation = Entities.UPDATE; |
| WebResource resource = service.path(operation.path).path(entityType).path(entityName); |
| if (effectiveTime != null) { |
| resource = resource.queryParam("effective", SchemaHelper.formatDateUTC(effectiveTime)); |
| } |
| ClientResponse clientResponse = resource |
| .header("Cookie", AUTH_COOKIE_EQ + authenticationToken) |
| .accept(operation.mimeType).type(MediaType.TEXT_XML) |
| .method(operation.method, ClientResponse.class, entityStream); |
| checkIfSuccessful(clientResponse); |
| return parseAPIResult(clientResponse); |
| } |
| |
| public String submitAndSchedule(String entityType, String filePath) |
| throws FalconCLIException { |
| |
| InputStream entityStream = getServletInputStream(filePath); |
| return sendEntityRequestWithObject(Entities.SUBMITandSCHEDULE, |
| entityType, entityStream, null); |
| } |
| |
| public String getStatus(String entityType, String entityName, String colo) |
| throws FalconCLIException { |
| |
| return sendEntityRequest(Entities.STATUS, entityType, entityName, colo); |
| } |
| |
| public String getDefinition(String entityType, String entityName) |
| throws FalconCLIException { |
| |
| return sendDefinitionRequest(Entities.DEFINITION, entityType, |
| entityName); |
| } |
| |
| public EntityList getDependency(String entityType, String entityName) |
| throws FalconCLIException { |
| return sendDependencyRequest(Entities.DEPENDENCY, entityType, entityName); |
| } |
| |
| //SUSPEND CHECKSTYLE CHECK ParameterNumberCheck |
| |
| public EntityList getEntityList(String entityType, String fields, String filterBy, String filterTags, |
| String orderBy, String sortOrder, |
| Integer offset, Integer numResults) throws FalconCLIException { |
| return sendListRequest(Entities.LIST, entityType, fields, filterBy, |
| filterTags, orderBy, sortOrder, offset, numResults); |
| } |
| |
| public String getEntitySummary(String entityType, String cluster, String start, String end, |
| String fields, String filterBy, String filterTags, |
| String orderBy, String sortOrder, |
| Integer offset, Integer numResults, Integer numInstances) |
| throws FalconCLIException { |
| return sendEntitySummaryRequest(Entities.SUMMARY, entityType, cluster, start, end, fields, filterBy, filterTags, |
| orderBy, sortOrder, offset, numResults, numInstances); |
| } |
| |
| public String getRunningInstances(String type, String entity, String colo, List<LifeCycle> lifeCycles, |
| String filterBy, String orderBy, String sortOrder, |
| Integer offset, Integer numResults) throws FalconCLIException { |
| |
| return sendInstanceRequest(Instances.RUNNING, type, entity, null, null, |
| null, null, colo, lifeCycles, filterBy, orderBy, sortOrder, offset, numResults); |
| } |
| |
| public String getStatusOfInstances(String type, String entity, |
| String start, String end, |
| String colo, List<LifeCycle> lifeCycles, String filterBy, |
| String orderBy, String sortOrder, |
| Integer offset, Integer numResults) throws FalconCLIException { |
| |
| return sendInstanceRequest(Instances.STATUS, type, entity, start, end, |
| null, null, colo, lifeCycles, filterBy, orderBy, sortOrder, offset, numResults); |
| } |
| |
| public String getSummaryOfInstances(String type, String entity, |
| String start, String end, |
| String colo, List<LifeCycle> lifeCycles) throws FalconCLIException { |
| |
| return sendInstanceRequest(Instances.SUMMARY, type, entity, start, end, |
| null, null, colo, lifeCycles); |
| } |
| |
| public String killInstances(String type, String entity, String start, |
| String end, String colo, String clusters, |
| String sourceClusters, List<LifeCycle> lifeCycles) |
| throws FalconCLIException, UnsupportedEncodingException { |
| |
| return sendInstanceRequest(Instances.KILL, type, entity, start, end, |
| getServletInputStream(clusters, sourceClusters, null), null, colo, lifeCycles); |
| } |
| |
| public String suspendInstances(String type, String entity, String start, |
| String end, String colo, String clusters, |
| String sourceClusters, List<LifeCycle> lifeCycles) |
| throws FalconCLIException, UnsupportedEncodingException { |
| |
| return sendInstanceRequest(Instances.SUSPEND, type, entity, start, end, |
| getServletInputStream(clusters, sourceClusters, null), null, colo, lifeCycles); |
| } |
| |
| public String resumeInstances(String type, String entity, String start, |
| String end, String colo, String clusters, |
| String sourceClusters, List<LifeCycle> lifeCycles) |
| throws FalconCLIException, UnsupportedEncodingException { |
| |
| return sendInstanceRequest(Instances.RESUME, type, entity, start, end, |
| getServletInputStream(clusters, sourceClusters, null), null, colo, lifeCycles); |
| } |
| |
| public String rerunInstances(String type, String entity, String start, |
| String end, String filePath, String colo, |
| String clusters, String sourceClusters, List<LifeCycle> lifeCycles) |
| throws FalconCLIException, IOException { |
| |
| StringBuilder buffer = new StringBuilder(); |
| if (filePath != null) { |
| BufferedReader in = null; |
| try { |
| in = new BufferedReader(new FileReader(filePath)); |
| |
| String str; |
| while ((str = in.readLine()) != null) { |
| buffer.append(str).append("\n"); |
| } |
| } finally { |
| IOUtils.closeQuietly(in); |
| } |
| } |
| String temp = (buffer.length() == 0) ? null : buffer.toString(); |
| return sendInstanceRequest(Instances.RERUN, type, entity, start, end, |
| getServletInputStream(clusters, sourceClusters, temp), null, colo, lifeCycles); |
| } |
| |
| public String rerunInstances(String type, String entity, String start, |
| String end, String colo, String clusters, String sourceClusters, |
| List<LifeCycle> lifeCycles) |
| throws FalconCLIException, UnsupportedEncodingException { |
| |
| return sendInstanceRequest(Instances.RERUN, type, entity, start, end, |
| getServletInputStream(clusters, sourceClusters, "oozie.wf.rerun.failnodes=true\n"), null, colo, |
| lifeCycles); |
| } |
| |
| public String getLogsOfInstances(String type, String entity, String start, |
| String end, String colo, String runId, |
| List<LifeCycle> lifeCycles, String filterBy, |
| String orderBy, String sortOrder, Integer offset, Integer numResults) |
| throws FalconCLIException { |
| |
| return sendInstanceRequest(Instances.LOG, type, entity, start, end, |
| null, runId, colo, lifeCycles, filterBy, orderBy, sortOrder, offset, numResults); |
| } |
| |
| public String getParamsOfInstance(String type, String entity, |
| String start, String colo, |
| String clusters, String sourceClusters, |
| List<LifeCycle> lifeCycles) |
| throws FalconCLIException, UnsupportedEncodingException { |
| |
| return sendInstanceRequest(Instances.PARAMS, type, entity, |
| start, null, null, null, colo, lifeCycles); |
| } |
| |
| public String getThreadDump() throws FalconCLIException { |
| return sendAdminRequest(AdminOperations.STACK); |
| } |
| |
| public String getVersion() throws FalconCLIException { |
| return sendAdminRequest(AdminOperations.VERSION); |
| } |
| |
| public int getStatus() throws FalconCLIException { |
| AdminOperations job = AdminOperations.VERSION; |
| ClientResponse clientResponse = service.path(job.path) |
| .header("Cookie", AUTH_COOKIE_EQ + authenticationToken) |
| .accept(job.mimeType).type(MediaType.TEXT_PLAIN) |
| .method(job.method, ClientResponse.class); |
| return clientResponse.getStatus(); |
| } |
| |
| /** |
| * Converts a InputStream into ServletInputStream. |
| * |
| * @param filePath - Path of file to stream |
| * @return ServletInputStream |
| * @throws FalconCLIException |
| */ |
| private InputStream getServletInputStream(String filePath) |
| throws FalconCLIException { |
| |
| if (filePath == null) { |
| return null; |
| } |
| InputStream stream; |
| try { |
| stream = new FileInputStream(filePath); |
| } catch (FileNotFoundException e) { |
| throw new FalconCLIException("File not found:", e); |
| } |
| return stream; |
| } |
| |
| private InputStream getServletInputStream(String clusters, |
| String sourceClusters, String properties) |
| throws FalconCLIException, UnsupportedEncodingException { |
| |
| InputStream stream; |
| StringBuilder buffer = new StringBuilder(); |
| if (clusters != null) { |
| buffer.append(FALCON_INSTANCE_ACTION_CLUSTERS).append('=').append(clusters).append('\n'); |
| } |
| if (sourceClusters != null) { |
| buffer.append(FALCON_INSTANCE_SOURCE_CLUSTERS).append('=').append(sourceClusters).append('\n'); |
| } |
| if (properties != null) { |
| buffer.append(properties); |
| } |
| stream = new ByteArrayInputStream(buffer.toString().getBytes()); |
| return (buffer.length() == 0) ? null : stream; |
| } |
| |
| private String sendEntityRequest(Entities entities, String entityType, |
| String entityName, String colo) throws FalconCLIException { |
| |
| WebResource resource = service.path(entities.path) |
| .path(entityType).path(entityName); |
| if (colo != null) { |
| resource = resource.queryParam("colo", colo); |
| } |
| ClientResponse clientResponse = resource |
| .header("Cookie", AUTH_COOKIE_EQ + authenticationToken) |
| .accept(entities.mimeType).type(MediaType.TEXT_XML) |
| .method(entities.method, ClientResponse.class); |
| |
| checkIfSuccessful(clientResponse); |
| |
| return parseAPIResult(clientResponse); |
| } |
| |
| private WebResource addParamsToResource(WebResource resource, |
| String start, String end, String runId, String colo, |
| String fields, String filterBy, String tags, |
| String orderBy, String sortOrder, |
| Integer offset, Integer numResults, Integer numInstances) { |
| |
| if (!StringUtils.isEmpty(fields)) { |
| resource = resource.queryParam("fields", fields); |
| } |
| if (!StringUtils.isEmpty(tags)) { |
| resource = resource.queryParam("tags", tags); |
| } |
| if (!StringUtils.isEmpty(filterBy)) { |
| resource = resource.queryParam("filterBy", filterBy); |
| } |
| if (!StringUtils.isEmpty(orderBy)) { |
| resource = resource.queryParam("orderBy", orderBy); |
| } |
| if (!StringUtils.isEmpty(sortOrder)) { |
| resource = resource.queryParam("sortOrder", sortOrder); |
| } |
| if (!StringUtils.isEmpty(start)) { |
| resource = resource.queryParam("start", start); |
| } |
| if (!StringUtils.isEmpty(end)) { |
| resource = resource.queryParam("end", end); |
| } |
| if (runId != null) { |
| resource = resource.queryParam("runid", runId); |
| } |
| if (colo != null) { |
| resource = resource.queryParam("colo", colo); |
| } |
| if (offset != null) { |
| resource = resource.queryParam("offset", offset.toString()); |
| } |
| if (numResults != null) { |
| resource = resource.queryParam("numResults", numResults.toString()); |
| } |
| if (numInstances != null) { |
| resource = resource.queryParam("numInstances", numInstances.toString()); |
| } |
| return resource; |
| |
| } |
| |
| private String sendEntitySummaryRequest(Entities entities, String entityType, String cluster, |
| String start, String end, |
| String fields, String filterBy, String filterTags, |
| String orderBy, String sortOrder, Integer offset, Integer numResults, |
| Integer numInstances) throws FalconCLIException { |
| WebResource resource; |
| if (StringUtils.isEmpty(cluster)) { |
| resource = service.path(entities.path).path(entityType); |
| } else { |
| resource = service.path(entities.path).path(entityType).path(cluster); |
| } |
| |
| resource = addParamsToResource(resource, start, end, null, null, |
| fields, filterBy, filterTags, |
| orderBy, sortOrder, |
| offset, numResults, numInstances); |
| |
| ClientResponse clientResponse = resource |
| .header("Cookie", AUTH_COOKIE_EQ + authenticationToken) |
| .accept(entities.mimeType).type(MediaType.TEXT_XML) |
| .method(entities.method, ClientResponse.class); |
| |
| checkIfSuccessful(clientResponse); |
| return parseProcessEntitySummaryResult(clientResponse); |
| } |
| //RESUME CHECKSTYLE CHECK ParameterNumberCheck |
| |
| private String sendDefinitionRequest(Entities entities, String entityType, |
| String entityName) throws FalconCLIException { |
| |
| ClientResponse clientResponse = service |
| .path(entities.path).path(entityType).path(entityName) |
| .header("Cookie", AUTH_COOKIE_EQ + authenticationToken) |
| .accept(entities.mimeType).type(MediaType.TEXT_XML) |
| .method(entities.method, ClientResponse.class); |
| |
| checkIfSuccessful(clientResponse); |
| return clientResponse.getEntity(String.class); |
| } |
| |
| private EntityList sendDependencyRequest(Entities entities, String entityType, |
| String entityName) throws FalconCLIException { |
| |
| ClientResponse clientResponse = service |
| .path(entities.path).path(entityType).path(entityName) |
| .header("Cookie", AUTH_COOKIE_EQ + authenticationToken) |
| .accept(entities.mimeType).type(MediaType.TEXT_XML) |
| .method(entities.method, ClientResponse.class); |
| |
| checkIfSuccessful(clientResponse); |
| |
| return parseEntityList(clientResponse); |
| } |
| |
| private String sendEntityRequestWithObject(Entities entities, String entityType, |
| Object requestObject, String colo) throws FalconCLIException { |
| WebResource resource = service.path(entities.path) |
| .path(entityType); |
| if (colo != null) { |
| resource = resource.queryParam("colo", colo); |
| } |
| ClientResponse clientResponse = resource |
| .header("Cookie", AUTH_COOKIE_EQ + authenticationToken) |
| .accept(entities.mimeType).type(MediaType.TEXT_XML) |
| .method(entities.method, ClientResponse.class, requestObject); |
| |
| checkIfSuccessful(clientResponse); |
| |
| return parseAPIResult(clientResponse); |
| } |
| |
| //SUSPEND CHECKSTYLE CHECK VisibilityModifierCheck |
| private String sendInstanceRequest(Instances instances, String type, |
| String entity, String start, String end, InputStream props, |
| String runid, String colo, |
| List<LifeCycle> lifeCycles) throws FalconCLIException { |
| return sendInstanceRequest(instances, type, entity, start, end, props, |
| runid, colo, lifeCycles, "", "", "", 0, DEFAULT_NUM_RESULTS); |
| } |
| |
| private String sendInstanceRequest(Instances instances, String type, String entity, |
| String start, String end, InputStream props, String runid, String colo, |
| List<LifeCycle> lifeCycles, String filterBy, String orderBy, String sortOrder, |
| Integer offset, Integer numResults) throws FalconCLIException { |
| checkType(type); |
| WebResource resource = service.path(instances.path).path(type) |
| .path(entity); |
| |
| resource = addParamsToResource(resource, start, end, runid, colo, |
| null, filterBy, null, orderBy, sortOrder, offset, numResults, null); |
| |
| if (lifeCycles != null) { |
| checkLifeCycleOption(lifeCycles, type); |
| for (LifeCycle lifeCycle : lifeCycles) { |
| resource = resource.queryParam("lifecycle", lifeCycle.toString()); |
| } |
| } |
| |
| ClientResponse clientResponse; |
| if (props == null) { |
| clientResponse = resource |
| .header("Cookie", AUTH_COOKIE_EQ + authenticationToken) |
| .accept(instances.mimeType) |
| .method(instances.method, ClientResponse.class); |
| } else { |
| clientResponse = resource |
| .header("Cookie", AUTH_COOKIE_EQ + authenticationToken) |
| .accept(instances.mimeType) |
| .method(instances.method, ClientResponse.class, props); |
| } |
| checkIfSuccessful(clientResponse); |
| |
| if (instances.name().equals("LOG")) { |
| return parseProcessInstanceResultLogs(clientResponse, runid); |
| } else if (instances.name().equals("SUMMARY")) { |
| return summarizeProcessInstanceResult(clientResponse); |
| } else { |
| return parseProcessInstanceResult(clientResponse); |
| } |
| |
| } |
| |
| //RESUME CHECKSTYLE CHECK VisibilityModifierCheck |
| |
| private void checkLifeCycleOption(List<LifeCycle> lifeCycles, String type) throws FalconCLIException { |
| if (lifeCycles != null && !lifeCycles.isEmpty()) { |
| EntityType entityType = EntityType.valueOf(type.toUpperCase().trim()); |
| for (LifeCycle lifeCycle : lifeCycles) { |
| if (entityType != lifeCycle.getTag().getType()) { |
| throw new FalconCLIException("Incorrect lifecycle: " + lifeCycle + "for given type: " + type); |
| } |
| } |
| } |
| } |
| |
| protected void checkType(String type) throws FalconCLIException { |
| if (type == null || type.isEmpty()) { |
| throw new FalconCLIException("entity type is empty"); |
| } else { |
| EntityType entityType = EntityType.valueOf(type.toUpperCase().trim()); |
| if (entityType == EntityType.CLUSTER) { |
| throw new FalconCLIException( |
| "Instance management functions don't apply to Cluster entities"); |
| } |
| } |
| } |
| |
| //SUSPEND CHECKSTYLE CHECK ParameterNumberCheck |
| private EntityList sendListRequest(Entities entities, String entityType, String fields, String filterBy, |
| String filterTags, String orderBy, String sortOrder, Integer offset, |
| Integer numResults) throws FalconCLIException { |
| WebResource resource = service.path(entities.path) |
| .path(entityType); |
| resource = addParamsToResource(resource, null, null, null, null, fields, filterBy, filterTags, |
| orderBy, sortOrder, offset, numResults, null); |
| |
| ClientResponse clientResponse = resource |
| .header("Cookie", AUTH_COOKIE_EQ + authenticationToken) |
| .accept(entities.mimeType).type(MediaType.TEXT_XML) |
| .method(entities.method, ClientResponse.class); |
| |
| checkIfSuccessful(clientResponse); |
| |
| return parseEntityList(clientResponse); |
| } |
| // RESUME CHECKSTYLE CHECK ParameterNumberCheck |
| |
| private String sendAdminRequest(AdminOperations job) throws FalconCLIException { |
| ClientResponse clientResponse = service.path(job.path) |
| .header("Cookie", AUTH_COOKIE_EQ + authenticationToken) |
| .accept(job.mimeType) |
| .type(job.mimeType) |
| .method(job.method, ClientResponse.class); |
| return parseStringResult(clientResponse); |
| } |
| |
| private String parseAPIResult(ClientResponse clientResponse) |
| throws FalconCLIException { |
| |
| APIResult result = clientResponse.getEntity(APIResult.class); |
| return result.getMessage(); |
| } |
| |
| private EntityList parseEntityList(ClientResponse clientResponse) |
| throws FalconCLIException { |
| |
| EntityList result = clientResponse.getEntity(EntityList.class); |
| if (result == null || result.getElements() == null) { |
| return null; |
| } |
| return result; |
| |
| } |
| |
| private String parseStringResult(ClientResponse clientResponse) |
| throws FalconCLIException { |
| |
| return clientResponse.getEntity(String.class); |
| } |
| |
| private String parseProcessEntitySummaryResult(ClientResponse clientResponse) { |
| EntitySummaryResult result = clientResponse.getEntity(EntitySummaryResult.class); |
| StringBuilder sb = new StringBuilder(); |
| String toAppend; |
| sb.append("Consolidated Status: ").append(result.getStatus()).append("\n"); |
| sb.append("\nEntity Summary Result :\n"); |
| if (result.getEntitySummaries() != null) { |
| for (EntitySummaryResult.EntitySummary entitySummary : result.getEntitySummaries()) { |
| |
| toAppend = entitySummary.toString(); |
| sb.append(toAppend).append("\n"); |
| } |
| } |
| sb.append("\nAdditional Information:\n"); |
| sb.append("Response: ").append(result.getMessage()); |
| sb.append("Request Id: ").append(result.getRequestId()); |
| return sb.toString(); |
| } |
| |
| private String summarizeProcessInstanceResult(ClientResponse clientResponse) { |
| InstancesSummaryResult result = clientResponse |
| .getEntity(InstancesSummaryResult.class); |
| StringBuilder sb = new StringBuilder(); |
| String toAppend; |
| |
| sb.append("Consolidated Status: ").append(result.getStatus()).append("\n"); |
| sb.append("\nInstances Summary:\n"); |
| |
| if (result.getInstancesSummary() != null) { |
| for (InstancesSummaryResult.InstanceSummary summary : result.getInstancesSummary()) { |
| toAppend = summary.getCluster() != null ? summary.getCluster() : "-"; |
| sb.append("Cluster: ").append(toAppend).append("\n"); |
| |
| sb.append("Status\t\tCount\n"); |
| sb.append("-------------------------\n"); |
| |
| for (Map.Entry<String, Long> entry : summary.getSummaryMap().entrySet()) { |
| sb.append(entry.getKey()).append("\t\t").append(entry.getValue()).append("\n"); |
| } |
| } |
| } |
| |
| sb.append("\nAdditional Information:\n"); |
| sb.append("Response: ").append(result.getMessage()); |
| sb.append("Request Id: ").append(result.getRequestId()); |
| return sb.toString(); |
| } |
| |
| private String parseProcessInstanceResult(ClientResponse clientResponse) { |
| InstancesResult result = clientResponse |
| .getEntity(InstancesResult.class); |
| StringBuilder sb = new StringBuilder(); |
| String toAppend; |
| |
| sb.append("Consolidated Status: ").append(result.getStatus()).append("\n"); |
| |
| sb.append("\nInstances:\n"); |
| sb.append("Instance\t\tCluster\t\tSourceCluster\t\tStatus\t\tStart\t\tEnd\t\tDetails\t\t\t\t\tLog\n"); |
| sb.append("-----------------------------------------------------------------------------------------------\n"); |
| if (result.getInstances() != null) { |
| for (InstancesResult.Instance instance : result.getInstances()) { |
| |
| toAppend = instance.getInstance() != null ? instance.getInstance() : "-"; |
| sb.append(toAppend).append("\t"); |
| |
| toAppend = instance.getCluster() != null ? instance.getCluster() : "-"; |
| sb.append(toAppend).append("\t"); |
| |
| toAppend = instance.getSourceCluster() != null ? instance.getSourceCluster() : "-"; |
| sb.append(toAppend).append("\t"); |
| |
| toAppend = (instance.getStatus() != null ? instance.getStatus().toString() : "-"); |
| sb.append(toAppend).append("\t"); |
| |
| toAppend = instance.getStartTime() != null |
| ? SchemaHelper.formatDateUTC(instance.getStartTime()) : "-"; |
| sb.append(toAppend).append("\t"); |
| |
| toAppend = instance.getEndTime() != null |
| ? SchemaHelper.formatDateUTC(instance.getEndTime()) : "-"; |
| sb.append(toAppend).append("\t"); |
| |
| toAppend = (!StringUtils.isEmpty(instance.getDetails())) |
| ? instance.getDetails() : "-"; |
| sb.append(toAppend).append("\t"); |
| |
| toAppend = instance.getLogFile() != null ? instance.getLogFile() : "-"; |
| sb.append(toAppend).append("\n"); |
| |
| if (instance.getWfParams() != null) { |
| Map<String, String> props = instance.getWfParams(); |
| sb.append("Workflow params").append("\n"); |
| for (Map.Entry<String, String> entry : props.entrySet()) { |
| sb.append(entry.getKey()).append("=").append(entry.getValue()).append("\n"); |
| } |
| sb.append("\n"); |
| } |
| |
| } |
| } |
| sb.append("\nAdditional Information:\n"); |
| sb.append("Response: ").append(result.getMessage()); |
| sb.append("Request Id: ").append(result.getRequestId()); |
| return sb.toString(); |
| } |
| |
| private String parseProcessInstanceResultLogs(ClientResponse clientResponse, String runid) { |
| InstancesResult result = clientResponse |
| .getEntity(InstancesResult.class); |
| StringBuilder sb = new StringBuilder(); |
| String toAppend; |
| |
| sb.append("Consolidated Status: ").append(result.getStatus()).append("\n"); |
| |
| sb.append("\nInstances:\n"); |
| sb.append("Instance\t\tCluster\t\tSourceCluster\t\tStatus\t\tRunID\t\t\tLog\n"); |
| sb.append("-----------------------------------------------------------------------------------------------\n"); |
| if (result.getInstances() != null) { |
| for (InstancesResult.Instance instance : result.getInstances()) { |
| |
| toAppend = (instance.getInstance() != null) ? instance.getInstance() : "-"; |
| sb.append(toAppend).append("\t"); |
| |
| toAppend = instance.getCluster() != null ? instance.getCluster() : "-"; |
| sb.append(toAppend).append("\t"); |
| |
| toAppend = instance.getSourceCluster() != null ? instance.getSourceCluster() : "-"; |
| sb.append(toAppend).append("\t"); |
| |
| toAppend = (instance.getStatus() != null ? instance.getStatus().toString() : "-"); |
| sb.append(toAppend).append("\t"); |
| |
| toAppend = (runid != null ? runid : "latest"); |
| sb.append(toAppend).append("\t"); |
| |
| toAppend = instance.getLogFile() != null ? instance.getLogFile() : "-"; |
| sb.append(toAppend).append("\n"); |
| |
| if (instance.actions != null) { |
| sb.append("actions:\n"); |
| for (InstancesResult.InstanceAction action : instance.actions) { |
| sb.append(" ").append(action.getAction()).append("\t"); |
| sb.append(action.getStatus()).append("\t").append(action.getLogFile()).append("\n"); |
| } |
| } |
| } |
| } |
| sb.append("\nAdditional Information:\n"); |
| sb.append("Response: ").append(result.getMessage()); |
| sb.append("Request Id: ").append(result.getRequestId()); |
| return sb.toString(); |
| } |
| |
| protected static enum GraphOperations { |
| |
| VERTICES("api/graphs/lineage/vertices", HttpMethod.GET, MediaType.APPLICATION_JSON), |
| EDGES("api/graphs/lineage/edges", HttpMethod.GET, MediaType.APPLICATION_JSON); |
| |
| private String path; |
| private String method; |
| private String mimeType; |
| |
| GraphOperations(String path, String method, String mimeType) { |
| this.path = path; |
| this.method = method; |
| this.mimeType = mimeType; |
| } |
| } |
| |
| public String getVertex(String id) throws FalconCLIException { |
| return sendGraphRequest(GraphOperations.VERTICES, id); |
| } |
| |
| public String getVertices(String key, String value) throws FalconCLIException { |
| return sendGraphRequest(GraphOperations.VERTICES, key, value); |
| } |
| |
| public String getVertexEdges(String id, String direction) throws FalconCLIException { |
| return sendGraphRequestForEdges(GraphOperations.VERTICES, id, direction); |
| } |
| |
| public String getEdge(String id) throws FalconCLIException { |
| return sendGraphRequest(GraphOperations.EDGES, id); |
| } |
| |
| private String sendGraphRequest(GraphOperations job, String id) throws FalconCLIException { |
| ClientResponse clientResponse = service.path(job.path) |
| .path(id) |
| .header("Cookie", AUTH_COOKIE_EQ + authenticationToken) |
| .accept(job.mimeType) |
| .type(job.mimeType) |
| .method(job.method, ClientResponse.class); |
| return parseStringResult(clientResponse); |
| } |
| |
| private String sendGraphRequest(GraphOperations job, String key, |
| String value) throws FalconCLIException { |
| ClientResponse clientResponse = service.path(job.path) |
| .queryParam("key", key) |
| .queryParam("value", value) |
| .header("Cookie", AUTH_COOKIE_EQ + authenticationToken) |
| .accept(job.mimeType) |
| .type(job.mimeType) |
| .method(job.method, ClientResponse.class); |
| return parseStringResult(clientResponse); |
| } |
| |
| private String sendGraphRequestForEdges(GraphOperations job, String id, |
| String direction) throws FalconCLIException { |
| ClientResponse clientResponse = service.path(job.path) |
| .path(id) |
| .path(direction) |
| .header("Cookie", AUTH_COOKIE_EQ + authenticationToken) |
| .accept(job.mimeType) |
| .type(job.mimeType) |
| .method(job.method, ClientResponse.class); |
| return parseStringResult(clientResponse); |
| } |
| |
| private void checkIfSuccessful(ClientResponse clientResponse) throws FalconCLIException { |
| Response.Status.Family statusFamily = clientResponse.getClientResponseStatus().getFamily(); |
| if (statusFamily != Response.Status.Family.SUCCESSFUL |
| && statusFamily != Response.Status.Family.INFORMATIONAL) { |
| throw FalconCLIException.fromReponse(clientResponse); |
| } |
| } |
| } |
