001    /**
002     * Licensed to the Apache Software Foundation (ASF) under one
003     * or more contributor license agreements.  See the NOTICE file
004     * distributed with this work for additional information
005     * regarding copyright ownership.  The ASF licenses this file
006     * to you under the Apache License, Version 2.0 (the
007     * "License"); you may not use this file except in compliance
008     * with the License.  You may obtain a copy of the License at
009     * 
010     *      http://www.apache.org/licenses/LICENSE-2.0
011     * 
012     * Unless required by applicable law or agreed to in writing, software
013     * distributed under the License is distributed on an "AS IS" BASIS,
014     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015     * See the License for the specific language governing permissions and
016     * limitations under the License.
017     */
018    package org.apache.oozie.command.wf;
019    
020    import org.apache.hadoop.conf.Configuration;
021    import org.apache.hadoop.fs.Path;
022    import org.apache.hadoop.fs.FileSystem;
023    import org.apache.oozie.WorkflowJobBean;
024    import org.apache.oozie.ErrorCode;
025    import org.apache.oozie.service.HadoopAccessorException;
026    import org.apache.oozie.service.JPAService;
027    import org.apache.oozie.service.WorkflowStoreService;
028    import org.apache.oozie.service.WorkflowAppService;
029    import org.apache.oozie.service.HadoopAccessorService;
030    import org.apache.oozie.service.Services;
031    import org.apache.oozie.service.DagXLogInfoService;
032    import org.apache.oozie.util.XLog;
033    import org.apache.oozie.util.ParamChecker;
034    import org.apache.oozie.util.XConfiguration;
035    import org.apache.oozie.util.XmlUtils;
036    import org.apache.oozie.command.CommandException;
037    import org.apache.oozie.executor.jpa.WorkflowJobInsertJPAExecutor;
038    import org.apache.oozie.service.ELService;
039    import org.apache.oozie.service.SchemaService;
040    import org.apache.oozie.store.StoreException;
041    import org.apache.oozie.workflow.WorkflowApp;
042    import org.apache.oozie.workflow.WorkflowException;
043    import org.apache.oozie.workflow.WorkflowInstance;
044    import org.apache.oozie.workflow.WorkflowLib;
045    import org.apache.oozie.util.ELEvaluator;
046    import org.apache.oozie.util.InstrumentUtils;
047    import org.apache.oozie.util.PropertiesUtils;
048    import org.apache.oozie.util.db.SLADbOperations;
049    import org.apache.oozie.service.SchemaService.SchemaName;
050    import org.apache.oozie.client.OozieClient;
051    import org.apache.oozie.client.WorkflowJob;
052    import org.apache.oozie.client.SLAEvent.SlaAppType;
053    import org.jdom.Element;
054    import org.jdom.Namespace;
055    
056    import java.util.Date;
057    import java.util.List;
058    import java.util.Map;
059    import java.util.Set;
060    import java.util.HashSet;
061    import java.io.IOException;
062    import java.net.URI;
063    
064    public class SubmitXCommand extends WorkflowXCommand<String> {
065        public static final String CONFIG_DEFAULT = "config-default.xml";
066    
067        private Configuration conf;
068        private String authToken;
069    
070        public SubmitXCommand(Configuration conf, String authToken) {
071            super("submit", "submit", 1);
072            this.conf = ParamChecker.notNull(conf, "conf");
073            this.authToken = ParamChecker.notEmpty(authToken, "authToken");
074        }
075    
076        private static final Set<String> DISALLOWED_DEFAULT_PROPERTIES = new HashSet<String>();
077        private static final Set<String> DISALLOWED_USER_PROPERTIES = new HashSet<String>();
078    
079        static {
080            String[] badUserProps = {PropertiesUtils.DAYS, PropertiesUtils.HOURS, PropertiesUtils.MINUTES,
081                    PropertiesUtils.KB, PropertiesUtils.MB, PropertiesUtils.GB, PropertiesUtils.TB, PropertiesUtils.PB,
082                    PropertiesUtils.RECORDS, PropertiesUtils.MAP_IN, PropertiesUtils.MAP_OUT, PropertiesUtils.REDUCE_IN,
083                    PropertiesUtils.REDUCE_OUT, PropertiesUtils.GROUPS};
084            PropertiesUtils.createPropertySet(badUserProps, DISALLOWED_USER_PROPERTIES);
085    
086            String[] badDefaultProps = {PropertiesUtils.HADOOP_USER, PropertiesUtils.HADOOP_UGI,
087                    WorkflowAppService.HADOOP_JT_KERBEROS_NAME, WorkflowAppService.HADOOP_NN_KERBEROS_NAME};
088            PropertiesUtils.createPropertySet(badUserProps, DISALLOWED_DEFAULT_PROPERTIES);
089            PropertiesUtils.createPropertySet(badDefaultProps, DISALLOWED_DEFAULT_PROPERTIES);
090        }
091    
092        @Override
093        protected String execute() throws CommandException {
094            InstrumentUtils.incrJobCounter(getName(), 1, getInstrumentation());
095            WorkflowAppService wps = Services.get().get(WorkflowAppService.class);
096            try {
097                XLog.Info.get().setParameter(DagXLogInfoService.TOKEN, conf.get(OozieClient.LOG_TOKEN));
098                WorkflowApp app = wps.parseDef(conf, authToken);
099                XConfiguration protoActionConf = wps.createProtoActionConf(conf, authToken, true);
100                WorkflowLib workflowLib = Services.get().get(WorkflowStoreService.class).getWorkflowLibWithNoDB();
101    
102                String user = conf.get(OozieClient.USER_NAME);
103                String group = conf.get(OozieClient.GROUP_NAME);
104                URI uri = new URI(conf.get(OozieClient.APP_PATH));
105                FileSystem fs = Services.get().get(HadoopAccessorService.class).createFileSystem(user,
106                        group, uri, new Configuration());
107    
108                Path configDefault = null;
109                // app path could be a directory
110                Path path = new Path(uri.getPath());
111                if (!fs.isFile(path)) {
112                    configDefault = new Path(path, SubmitCommand.CONFIG_DEFAULT);
113                } else {
114                    configDefault = new Path(path.getParent(), SubmitCommand.CONFIG_DEFAULT);
115                }
116    
117                if (fs.exists(configDefault)) {
118                    try {
119                        Configuration defaultConf = new XConfiguration(fs.open(configDefault));
120                        PropertiesUtils.checkDisallowedProperties(defaultConf, DISALLOWED_DEFAULT_PROPERTIES);
121                        XConfiguration.injectDefaults(defaultConf, conf);
122                    }
123                    catch (IOException ex) {
124                        throw new IOException("default configuration file, " + ex.getMessage(), ex);
125                    }
126                }
127    
128                PropertiesUtils.checkDisallowedProperties(conf, DISALLOWED_USER_PROPERTIES);
129    
130                // Resolving all variables in the job properties.
131                // This ensures the Hadoop Configuration semantics is preserved.
132                XConfiguration resolvedVarsConf = new XConfiguration();
133                for (Map.Entry<String, String> entry : conf) {
134                    resolvedVarsConf.set(entry.getKey(), conf.get(entry.getKey()));
135                }
136                conf = resolvedVarsConf;
137    
138                WorkflowInstance wfInstance;
139                try {
140                    wfInstance = workflowLib.createInstance(app, conf);
141                }
142                catch (WorkflowException e) {
143                    throw new StoreException(e);
144                }
145    
146                Configuration conf = wfInstance.getConf();
147                // System.out.println("WF INSTANCE CONF:");
148                // System.out.println(XmlUtils.prettyPrint(conf).toString());
149    
150                WorkflowJobBean workflow = new WorkflowJobBean();
151                workflow.setId(wfInstance.getId());
152                workflow.setAppName(app.getName());
153                workflow.setAppPath(conf.get(OozieClient.APP_PATH));
154                workflow.setConf(XmlUtils.prettyPrint(conf).toString());
155                workflow.setProtoActionConf(protoActionConf.toXmlString());
156                workflow.setCreatedTime(new Date());
157                workflow.setLastModifiedTime(new Date());
158                workflow.setLogToken(conf.get(OozieClient.LOG_TOKEN, ""));
159                workflow.setStatus(WorkflowJob.Status.PREP);
160                workflow.setRun(0);
161                workflow.setUser(conf.get(OozieClient.USER_NAME));
162                workflow.setGroup(conf.get(OozieClient.GROUP_NAME));
163                workflow.setAuthToken(authToken);
164                workflow.setWorkflowInstance(wfInstance);
165                workflow.setExternalId(conf.get(OozieClient.EXTERNAL_ID));
166    
167                //setLogInfo(workflow);
168                Element wfElem = XmlUtils.parseXml(app.getDefinition());
169                ELEvaluator evalSla = createELEvaluatorForGroup(conf, "wf-sla-submit");
170                String jobSlaXml = verifySlaElements(wfElem, evalSla);
171                writeSLARegistration(jobSlaXml, workflow.getId(), workflow.getUser(), workflow.getGroup(), LOG);
172                workflow.setSlaXml(jobSlaXml);
173                // System.out.println("SlaXml :"+ slaXml);
174    
175                //store.insertWorkflow(workflow);
176                JPAService jpaService = Services.get().get(JPAService.class);
177                if (jpaService != null) {
178                    jpaService.execute(new WorkflowJobInsertJPAExecutor(workflow));
179                }
180                else {
181                    LOG.error(ErrorCode.E0610);
182                    return null;
183                }
184    
185                return workflow.getId();
186            }
187            catch (WorkflowException ex) {
188                throw new CommandException(ex);
189            }
190            catch (HadoopAccessorException ex) {
191                throw new CommandException(ex);
192            }
193            catch (Exception ex) {
194                throw new CommandException(ErrorCode.E0803, ex);
195            }
196        }
197    
198        private String verifySlaElements(Element eWfJob, ELEvaluator evalSla) throws CommandException {
199            String jobSlaXml = "";
200            // Validate WF job
201            Element eSla = eWfJob.getChild("info", Namespace.getNamespace(SchemaService.SLA_NAME_SPACE_URI));
202            if (eSla != null) {
203                jobSlaXml = resolveSla(eSla, evalSla);
204            }
205    
206            // Validate all actions
207            for (Element action : (List<Element>) eWfJob.getChildren("action", eWfJob.getNamespace())) {
208                eSla = action.getChild("info", Namespace.getNamespace(SchemaService.SLA_NAME_SPACE_URI));
209                if (eSla != null) {
210                    resolveSla(eSla, evalSla);
211                }
212            }
213            return jobSlaXml;
214        }
215    
216        private void writeSLARegistration(String slaXml, String id, String user, String group, XLog log)
217                throws CommandException {
218            try {
219                if (slaXml != null && slaXml.length() > 0) {
220                    Element eSla = XmlUtils.parseXml(slaXml);
221                    SLADbOperations.writeSlaRegistrationEvent(eSla, id, SlaAppType.WORKFLOW_JOB, user, group, log);
222                }
223            }
224            catch (Exception e) {
225                e.printStackTrace();
226                throw new CommandException(ErrorCode.E1007, "workflow " + id, e);
227            }
228        }
229    
230        /**
231         * Resolve variables in sla xml element.
232         *
233         * @param eSla sla xml element
234         * @param evalSla sla evaluator
235         * @return sla xml string after evaluation
236         * @throws CommandException
237         */
238        public static String resolveSla(Element eSla, ELEvaluator evalSla) throws CommandException {
239            // EL evaluation
240            String slaXml = XmlUtils.prettyPrint(eSla).toString();
241            try {
242                slaXml = XmlUtils.removeComments(slaXml);
243                slaXml = evalSla.evaluate(slaXml, String.class);
244                XmlUtils.validateData(slaXml, SchemaName.SLA_ORIGINAL);
245                return slaXml;
246            }
247            catch (Exception e) {
248                throw new CommandException(ErrorCode.E1004, "Validation erro :" + e.getMessage(), e);
249            }
250        }
251    
252        /**
253         * Create an EL evaluator for a given group.
254         *
255         * @param conf configuration variable
256         * @param group group variable
257         * @return the evaluator created for the group
258         */
259        public static ELEvaluator createELEvaluatorForGroup(Configuration conf, String group) {
260            ELEvaluator eval = Services.get().get(ELService.class).createEvaluator(group);
261            for (Map.Entry<String, String> entry : conf) {
262                eval.setVariable(entry.getKey(), entry.getValue());
263            }
264            return eval;
265        }
266    
267        @Override
268        protected String getEntityKey() {
269            return null;
270        }
271    
272        @Override
273        protected boolean isLockRequired() {
274            return false;
275        }
276    
277        @Override
278        protected void loadState() {
279    
280        }
281    
282        @Override
283        protected void verifyPrecondition() throws CommandException {
284    
285        }
286    
287    }