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
019package org.apache.oozie.command.wf;
020
021import org.apache.oozie.client.OozieClient;
022import org.apache.oozie.WorkflowActionBean;
023import org.apache.oozie.WorkflowJobBean;
024import org.apache.oozie.command.CommandException;
025import org.apache.oozie.command.PreconditionException;
026import org.apache.oozie.service.ConfigurationService;
027import org.apache.oozie.util.LogUtils;
028import org.apache.oozie.util.ParamChecker;
029import org.apache.oozie.util.XLog;
030
031import java.io.IOException;
032import java.net.HttpURLConnection;
033import java.net.URL;
034
035public class WorkflowNotificationXCommand extends WorkflowXCommand<Void> {
036
037    public static final String NOTIFICATION_URL_CONNECTION_TIMEOUT_KEY = "oozie.notification.url.connection.timeout";
038
039    private static final String STATUS_PATTERN = "\\$status";
040    private static final String JOB_ID_PATTERN = "\\$jobId";
041    private static final String NODE_NAME_PATTERN = "\\$nodeName";
042
043    private String url;
044    private String id;
045
046    //this variable is package private only for test purposes
047    int retries = 0;
048
049    public WorkflowNotificationXCommand(WorkflowJobBean workflow) {
050        super("job.notification", "job.notification", 0);
051        ParamChecker.notNull(workflow, "workflow");
052        id = workflow.getId();
053        url = workflow.getWorkflowInstance().getConf().get(OozieClient.WORKFLOW_NOTIFICATION_URL);
054        if (url != null) {
055            url = url.replaceAll(JOB_ID_PATTERN, workflow.getId());
056            url = url.replaceAll(STATUS_PATTERN, workflow.getStatus().toString());
057        }
058    }
059
060    public WorkflowNotificationXCommand(WorkflowJobBean workflow, WorkflowActionBean action) {
061        super("action.notification", "job.notification", 0);
062        ParamChecker.notNull(workflow, "workflow");
063        ParamChecker.notNull(action, "action");
064        id = action.getId();
065        url = workflow.getWorkflowInstance().getConf().get(OozieClient.ACTION_NOTIFICATION_URL);
066        if (url != null) {
067            url = url.replaceAll(JOB_ID_PATTERN, workflow.getId());
068            url = url.replaceAll(NODE_NAME_PATTERN, action.getName());
069            if (action.isComplete()) {
070                url = url.replaceAll(STATUS_PATTERN, "T:" + action.getTransition());
071            }
072            else {
073                url = url.replaceAll(STATUS_PATTERN, "S:" + action.getStatus().toString());
074            }
075        }
076    }
077
078    @Override
079    protected void setLogInfo() {
080        LogUtils.setLogInfo(id);
081    }
082
083    @Override
084    protected boolean isLockRequired() {
085        return false;
086    }
087
088    @Override
089    public String getEntityKey() {
090        return url;
091    }
092
093    @Override
094    protected void loadState() throws CommandException {
095    }
096
097    @Override
098    protected void verifyPrecondition() throws CommandException, PreconditionException {
099    }
100
101    @Override
102    protected Void execute() throws CommandException {
103        if (url != null) {
104            int timeout = ConfigurationService.getInt(NOTIFICATION_URL_CONNECTION_TIMEOUT_KEY);
105            try {
106                URL url = new URL(this.url);
107                HttpURLConnection urlConn = (HttpURLConnection) url.openConnection();
108                urlConn.setConnectTimeout(timeout);
109                urlConn.setReadTimeout(timeout);
110                if (urlConn.getResponseCode() != HttpURLConnection.HTTP_OK) {
111                    handleRetry();
112                }
113            }
114            catch (IOException ex) {
115                handleRetry();
116            }
117        }
118        return null;
119    }
120
121    private void handleRetry() {
122        if (retries < 3) {
123            retries++;
124            this.resetUsed();
125            queue(this, 60 * 1000);
126        }
127        else {
128            LOG.warn(XLog.OPS, "could not send notification [{0}]", url);
129        }
130    }
131
132    public String getUrl() {
133        return url;
134    }
135
136}