diff --git ivy/ivy.xml ivy/ivy.xml index 2c7184e..3a83570 100644 --- ivy/ivy.xml +++ ivy/ivy.xml @@ -89,8 +89,10 @@ - - + + diff --git src/java/org/apache/nutch/api/APIInfoResource.java src/java/org/apache/nutch/api/APIInfoResource.java deleted file mode 100644 index 69dd944..0000000 --- src/java/org/apache/nutch/api/APIInfoResource.java +++ /dev/null @@ -1,40 +0,0 @@ -/******************************************************************************* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - ******************************************************************************/ -package org.apache.nutch.api; - -import java.io.IOException; -import java.util.ArrayList; -import java.util.List; - -import org.restlet.resource.Get; -import org.restlet.resource.ServerResource; - -public class APIInfoResource extends ServerResource { - private static final List info = new ArrayList(); - - static { - info.add(new String[]{AdminResource.PATH, AdminResource.DESCR}); - info.add(new String[]{ConfResource.PATH, ConfResource.DESCR}); - info.add(new String[]{DbResource.PATH, DbResource.DESCR}); - info.add(new String[]{JobResource.PATH, JobResource.DESCR}); - } - - @Get("json") - public List retrieve() throws IOException { - return info; - } -} diff --git src/java/org/apache/nutch/api/AdminResource.java src/java/org/apache/nutch/api/AdminResource.java deleted file mode 100644 index 77604ba..0000000 --- src/java/org/apache/nutch/api/AdminResource.java +++ /dev/null @@ -1,94 +0,0 @@ -/******************************************************************************* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - ******************************************************************************/ -package org.apache.nutch.api; - -import java.util.Map; - -import org.apache.commons.lang.BooleanUtils; -import org.apache.commons.lang.StringUtils; -import org.apache.nutch.api.JobStatus.State; -import org.restlet.resource.Get; -import org.restlet.resource.ServerResource; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.google.common.collect.Maps; - -public class AdminResource extends ServerResource { - private static final int SHUTDOWN_DELAY = 1000; - - private static final Logger LOG = LoggerFactory.getLogger(AdminResource.class); - - public static final String PATH = "admin"; - public static final String DESCR = "Service admin actions"; - - @Get("json") - public Object execute() throws Exception { - Map attributes = getRequestAttributes(); - String cmd = (String) attributes.get(Params.CMD); - - if (StringUtils.equalsIgnoreCase("status", cmd)) { - return getNutchStatus(); - } - - if (StringUtils.equalsIgnoreCase("stop", cmd)) { - boolean force = BooleanUtils.toBoolean(getQuery().getFirstValue(Params.FORCE)); - return stopServer(force); - } - return "Unknown command " + cmd; - } - - private String stopServer(boolean force) throws Exception { - if (!canStopServer(force)) { - LOG.info("Command 'stop' denied due to unfinished jobs"); - return "can't stop now"; - } - - Thread t = new Thread() { - public void run() { - try { - Thread.sleep(SHUTDOWN_DELAY); - NutchApp.server.stop(false); - LOG.info("Service stopped."); - } catch (Exception e) { - LOG.error("Error stopping", e); - } - } - }; - t.setDaemon(true); - t.start(); - LOG.info("Service shutting down..."); - return "stopping"; - } - - private boolean canStopServer(boolean force) throws Exception { - return force || NutchApp.server.canStop(); - } - - private Map getNutchStatus() throws Exception { - Map res = Maps.newHashMap(); - res.put("started", NutchApp.started); - - Map jobs = Maps.newHashMap(); - jobs.put("all", NutchApp.jobMgr.list(null, State.ANY)); - jobs.put("running", NutchApp.jobMgr.list(null, State.RUNNING)); - res.put("jobs", jobs); - res.put("confs", NutchApp.confMgr.list()); - - return res; - } -} diff --git src/java/org/apache/nutch/api/ConfManager.java src/java/org/apache/nutch/api/ConfManager.java index bfa8e9b..34896f3 100644 --- src/java/org/apache/nutch/api/ConfManager.java +++ src/java/org/apache/nutch/api/ConfManager.java @@ -20,18 +20,19 @@ import java.util.Map; import java.util.Set; import org.apache.hadoop.conf.Configuration; +import org.apache.nutch.api.model.request.NutchConfig; public interface ConfManager { - public Set list() throws Exception; - + public Set list(); + public Configuration get(String confId); - - public Map getAsMap(String confId); - + + public Map getAsMap(String confId); + public void delete(String confId); - - public void create(String confId, Map props, boolean force) throws Exception; - - public void setProperty(String confId, String propName, String propValue) throws Exception; + + public void setProperty(String confId, String propName, String propValue); + + public String create(NutchConfig nutchConfig); } diff --git src/java/org/apache/nutch/api/ConfResource.java src/java/org/apache/nutch/api/ConfResource.java deleted file mode 100644 index 8be8333..0000000 --- src/java/org/apache/nutch/api/ConfResource.java +++ /dev/null @@ -1,102 +0,0 @@ -/******************************************************************************* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - ******************************************************************************/ -package org.apache.nutch.api; - -import java.util.Map; -import java.util.concurrent.atomic.AtomicInteger; - -import org.restlet.resource.Delete; -import org.restlet.resource.Get; -import org.restlet.resource.Post; -import org.restlet.resource.Put; -import org.restlet.resource.ServerResource; - -public class ConfResource extends ServerResource { - - public static final String PATH = "confs"; - public static final String DESCR = "Configuration manager"; - public static final String DEFAULT_CONF = "default"; - - private AtomicInteger seqId = new AtomicInteger(); - - @Get("json") - public Object retrieve() throws Exception { - String id = (String)getRequestAttributes().get(Params.CONF_ID); - if (id == null) { - return NutchApp.confMgr.list(); - } else { - String prop = (String)getRequestAttributes().get(Params.PROP_NAME); - if (prop == null) { - return NutchApp.confMgr.getAsMap(id); - } else { - Map cfg = NutchApp.confMgr.getAsMap(id); - if (cfg == null) { - return null; - } else { - return cfg.get(prop); - } - } - } - } - - @SuppressWarnings("unchecked") - @Put("json") - public String create(Map args) throws Exception { - System.out.println("args=" + args); - String id = (String)args.get(Params.CONF_ID); - if (id == null) { - id = String.valueOf(seqId.incrementAndGet()); - } - Object temp = args.get(Params.PROPS); - Map props = null; - if(temp instanceof Map) - props = (Map) temp; - - Boolean force = (Boolean)args.get(Params.FORCE); - boolean f = force != null ? force : false; - NutchApp.confMgr.create(id, props, f); - return id; - } - - @Post("json") - public void update(Map args) throws Exception { - String id = (String)args.get(Params.CONF_ID); - if (id == null) id = (String)getRequestAttributes().get(Params.CONF_ID); - if (id == null) { - throw new Exception("Missing config id"); - } - String prop = (String)args.get(Params.PROP_NAME); - if (prop == null) prop = (String)getRequestAttributes().get(Params.PROP_NAME); - if (prop == null) { - throw new Exception("Missing property name prop"); - } - String value = (String)args.get(Params.PROP_VALUE); - if (value == null) { - throw new Exception("Missing property value"); - } - NutchApp.confMgr.setProperty(id, prop, value); - } - - @Delete - public void remove() throws Exception { - String id = (String)getRequestAttributes().get(Params.CONF_ID); - if (id == null) { - throw new Exception("Missing config id"); - } - NutchApp.confMgr.delete(id); - } -} diff --git src/java/org/apache/nutch/api/DbReader.java src/java/org/apache/nutch/api/DbReader.java deleted file mode 100644 index 68cc9ac..0000000 --- src/java/org/apache/nutch/api/DbReader.java +++ /dev/null @@ -1,237 +0,0 @@ -/******************************************************************************* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - ******************************************************************************/ -package org.apache.nutch.api; - -import org.apache.avro.Schema; -import org.apache.avro.util.Utf8; -import org.apache.gora.query.Query; -import org.apache.gora.query.Result; -import org.apache.gora.store.DataStore; -import org.apache.hadoop.conf.Configuration; -import org.apache.nutch.metadata.Nutch; -import org.apache.nutch.parse.ParseStatusUtils; -import org.apache.nutch.protocol.ProtocolStatusUtils; -import org.apache.nutch.storage.*; -import org.apache.nutch.util.Bytes; -import org.apache.nutch.util.NutchJob; -import org.apache.nutch.util.StringUtil; -import org.apache.nutch.util.TableUtil; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.IOException; -import java.nio.ByteBuffer; -import java.util.*; -import java.util.Map.Entry; - -public class DbReader { - private static final Logger LOG = LoggerFactory.getLogger(DbReader.class); - - DataStore store; - Configuration conf; - - public DbReader(Configuration conf, String crawlId) { - conf = new Configuration(conf); - if (crawlId != null) { - conf.set(Nutch.CRAWL_ID_KEY, crawlId); - } - try { - store = StorageUtils.createWebStore(conf, String.class, WebPage.class); - } catch (Exception e) { - e.printStackTrace(); - store = null; - } - } - - public Iterator> iterator(String[] fields, String startKey, String endKey, - String batchId) throws Exception { - Query q = store.newQuery(); - String[] qFields = fields; - if (fields != null) { - HashSet flds = new HashSet(Arrays.asList(fields)); - // remove "url" - flds.remove("url"); - if (flds.size() > 0) { - qFields = flds.toArray(new String[flds.size()]); - } else { - qFields = null; - } - } - q.setFields(qFields); - if (startKey != null) { - q.setStartKey(startKey); - if (endKey != null) { - q.setEndKey(endKey); - } - } - Result res = store.execute(q); - // XXX we should add the filtering capability to Query - return new DbIterator(res, fields, batchId); - } - - public void close() throws IOException { - if (store != null) { - store.close(); - } - } - - private class DbIterator implements Iterator> { - private Result res; - private boolean hasNext; - private String url; - private WebPage page; - private Utf8 batchId; - private TreeSet fields; - - DbIterator(Result res, String[] fields, String batchId) throws IOException { - this.res = res; - if (batchId != null) { - this.batchId = new Utf8(batchId); - } - if (fields != null) { - this.fields = new TreeSet(Arrays.asList(fields)); - } - try { - advance(); - } catch (Exception e){ - e.printStackTrace(); - } - } - - private void advance() throws Exception, IOException { - hasNext = res.next(); - if (hasNext && batchId != null) { - do { - WebPage page = res.get(); - Utf8 mark = Mark.UPDATEDB_MARK.checkMark(page); - if (NutchJob.shouldProcess(mark, batchId)) { - return; - } else { - if (LOG.isDebugEnabled()) { - LOG.debug("Skipping " + - TableUtil.unreverseUrl(res.getKey()) + "; different batch id"); - } - hasNext = res.next(); - } - } while (hasNext); - } - } - - public boolean hasNext() { - return hasNext; - } - - public Map next() { - url = res.getKey(); - page = WebPage.newBuilder(res.get()).build(); - try { - advance(); - if (!hasNext) { - res.close(); - } - } catch (IOException e) { - e.printStackTrace(); - hasNext = false; - return null; - } catch (Exception e) { - e.printStackTrace(); - hasNext = false; - return null; - } - return pageAsMap(url, page); - } - - @SuppressWarnings("unchecked") - private Map pageAsMap(String url, WebPage page) { - HashMap res = new HashMap(); - if (fields == null || fields.contains("url")) { - res.put("url", TableUtil.unreverseUrl(url)); - } - List pfields = page.getSchema().getFields(); - TreeSet flds = null; - if (fields != null) { - flds = (TreeSet) fields.clone(); - } else { - flds = new TreeSet(pfields); - } - flds.retainAll(Arrays.asList(pfields)); - for (Schema.Field f : flds) { - int idx = f.pos(); - if (idx < 0) { - continue; - } - Object val = page.get(idx); - if (val == null) { - continue; - } - if ("metadata".equals(f)) { - Map metadata = page.getMetadata(); - Map simpleMeta = new HashMap(); - if (metadata != null) { - Iterator> iterator = metadata.entrySet() - .iterator(); - while (iterator.hasNext()) { - Entry entry = iterator.next(); - simpleMeta.put(entry.getKey().toString(), - Bytes.toStringBinary(entry.getValue())); - } - } - res.put(f.name(), simpleMeta); - } else if ("protocolStatus".equals(f)) { - ProtocolStatus ps = page.getProtocolStatus(); - res.put(f.name(), ProtocolStatusUtils.toString(ps)); - } else if ("parseStatus".equals(f)) { - ParseStatus ps = page.getParseStatus(); - res.put(f.name(), ParseStatusUtils.toString(ps)); - } else if ("signature".equals(f)) { - ByteBuffer bb = page.getSignature(); - res.put(f.name(), StringUtil.toHexString(bb)); - } else if ("content".equals(f)) { - ByteBuffer bb = page.getContent(); - res.put(f.name(), Bytes.toStringBinary(bb)); - } else if ("markers".equals(f)) { - res.put(f.name(), convertMap(page.getMarkers())); - } else if ("inlinks".equals(f)) { - res.put(f.name(), convertMap(page.getInlinks())); - } else if ("outlinks".equals(f)) { - res.put(f.name(), convertMap(page.getOutlinks())); - } else { - if (val instanceof Utf8) { - val = val.toString(); - } else if (val instanceof ByteBuffer) { - val = Bytes.toStringBinary((ByteBuffer)val); - } - res.put(f.name(), val); - } - } - return res; - } - - private Map convertMap(Map map) { - Map res = new HashMap(); - for (Object o : map.entrySet()) { - Entry e = (Entry)o; - res.put(e.getKey().toString(), e.getValue().toString()); - } - return res; - } - - public void remove() { - throw new UnsupportedOperationException(); - } - } -} diff --git src/java/org/apache/nutch/api/DbResource.java src/java/org/apache/nutch/api/DbResource.java deleted file mode 100644 index c4927f4..0000000 --- src/java/org/apache/nutch/api/DbResource.java +++ /dev/null @@ -1,142 +0,0 @@ -/******************************************************************************* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - ******************************************************************************/ -package org.apache.nutch.api; - -import java.io.IOException; -import java.io.OutputStream; -import java.net.MalformedURLException; -import java.util.Iterator; -import java.util.Map; -import java.util.WeakHashMap; - -import org.apache.nutch.util.TableUtil; -import org.restlet.data.Form; -import org.restlet.data.MediaType; -import org.restlet.ext.jackson.JacksonConverter; -import org.restlet.representation.OutputRepresentation; -import org.restlet.representation.Representation; -import org.restlet.representation.Variant; -import org.restlet.resource.ResourceException; -import org.restlet.resource.ServerResource; - -public class DbResource extends ServerResource { - public static final String PATH = "db"; - public static final String DESCR = "DB data streaming"; - - static JacksonConverter cnv = new JacksonConverter(); - WeakHashMap readers = new WeakHashMap(); - - @Override - protected void doInit() throws ResourceException { - super.doInit(); - getVariants().add(new Variant(MediaType.APPLICATION_JSON)); - } - - @Override - protected Representation get(final Variant variant) throws ResourceException { - String startKey = null; - String endKey = null; - String rStartKey = null; // reversed keys - String rEndKey = null; - String[] fields = null; - String batchId = null; - String confId = ConfResource.DEFAULT_CONF; - Form form = getQuery(); - if (form != null) { - startKey = form.getFirstValue("start"); - endKey = form.getFirstValue("end"); - rStartKey = form.getFirstValue("rstart"); - rEndKey = form.getFirstValue("rend"); - if (rStartKey != null || rEndKey != null) { - startKey = rStartKey; - endKey = rEndKey; - } else { - if (startKey != null) { - try { - startKey = TableUtil.reverseUrl(startKey); - } catch (MalformedURLException e) { /*ignore */}; - } - if (endKey != null) { - try { - endKey = TableUtil.reverseUrl(endKey); - } catch (MalformedURLException e) { /*ignore */}; - } - } - batchId = form.getFirstValue("batch"); - String flds = form.getFirstValue("fields"); - if (flds != null && flds.trim().length() > 0) { - flds = flds.replaceAll("\\s+", ""); - fields = flds.split(","); - } - } - DbReader reader; - synchronized (readers) { - reader = readers.get(confId); - if (reader == null) { - reader = new DbReader(NutchApp.confMgr.get(confId), null); - readers.put(confId, reader); - } - } - Representation res = new DbRepresentation(this, variant, reader, fields, - startKey, endKey, batchId); - return res; - } - - private static class DbRepresentation extends OutputRepresentation { - private DbReader r; - private Variant variant; - private String[] fields; - private String startKey, endKey, batchId; - private DbResource resource; - - public DbRepresentation(DbResource resource, Variant variant, DbReader reader, - String[] fields, String startKey, String endKey, String batchId) { - super(variant.getMediaType()); - this.resource = resource; - this.r = reader; - this.variant = variant; - this.fields = fields; - this.startKey = startKey; - this.endKey = endKey; - this.batchId = batchId; - } - - @Override - public void write(OutputStream out) throws IOException { - try { - out.write('['); - Iterator> it = r.iterator(fields, startKey, endKey, batchId); - boolean first = true; - while (it.hasNext()) { - if (!first) { - out.write(','); - } else { - first = false; - } - Map item = it.next(); - Representation repr = cnv.toRepresentation(item, variant, resource); - repr.write(out); - out.flush(); - repr.release(); - } - out.write(']'); - } catch (Exception e) { - throw new IOException("DbReader.iterator failed", e); - } - } - } -} diff --git src/java/org/apache/nutch/api/JobManager.java src/java/org/apache/nutch/api/JobManager.java index 19941d2..aa25cc1 100644 --- src/java/org/apache/nutch/api/JobManager.java +++ src/java/org/apache/nutch/api/JobManager.java @@ -16,22 +16,25 @@ ******************************************************************************/ package org.apache.nutch.api; -import java.util.List; -import java.util.Map; -import org.apache.nutch.api.JobStatus.State; +import java.util.Collection; + +import org.apache.nutch.api.model.request.JobConfig; +import org.apache.nutch.api.model.response.JobInfo; +import org.apache.nutch.api.model.response.JobInfo.State; public interface JobManager { - - public static enum JobType {INJECT, GENERATE, FETCH, PARSE, UPDATEDB, INDEX, READDB, CLASS}; - - public List list(String crawlId, State state) throws Exception; - - public JobStatus get(String crawlId, String id) throws Exception; - - public String create(String crawlId, JobType type, String confId, - Map args) throws Exception; - - public boolean abort(String crawlId, String id) throws Exception; - - public boolean stop(String crawlId, String id) throws Exception; + + public static enum JobType { + INJECT, GENERATE, FETCH, PARSE, UPDATEDB, INDEX, READDB, CLASS + }; + + public Collection list(String crawlId, State state); + + public JobInfo get(String crawlId, String id); + + public String create(JobConfig jobConfig); + + public boolean abort(String crawlId, String id); + + public boolean stop(String crawlId, String id); } diff --git src/java/org/apache/nutch/api/JobResource.java src/java/org/apache/nutch/api/JobResource.java deleted file mode 100644 index 33ac4f2..0000000 --- src/java/org/apache/nutch/api/JobResource.java +++ /dev/null @@ -1,89 +0,0 @@ -/******************************************************************************* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - ******************************************************************************/ -package org.apache.nutch.api; - -import java.util.Map; - -import org.apache.nutch.api.JobManager.JobType; -import org.apache.nutch.api.JobStatus.State; -import org.restlet.data.Form; -import org.restlet.resource.Get; -import org.restlet.resource.Put; -import org.restlet.resource.ServerResource; - -public class JobResource extends ServerResource { - public static final String PATH = "jobs"; - public static final String DESCR = "Job manager"; - - @Get("json") - public Object retrieve() throws Exception { - String cid = null; - String jid = null; - String cmd = null; - Form form = getQuery(); - cid = (String)getRequestAttributes().get(Params.CRAWL_ID); - jid = (String)getRequestAttributes().get(Params.JOB_ID); - cmd = (String)getRequestAttributes().get(Params.CMD); - if (form != null) { - String v = form.getFirstValue(Params.CRAWL_ID); - if (v != null) cid = v; - v = form.getFirstValue(Params.JOB_ID); - if (v != null) jid = v; - v = form.getFirstValue(Params.CMD); - if (v != null) cmd = v; - } - if (jid == null) { - return NutchApp.jobMgr.list(cid, State.ANY); - } else { - // handle stop / abort / get - if (cmd == null) { - return NutchApp.jobMgr.get(cid, jid); - } - if (cmd.equals(Params.JOB_CMD_STOP)) { - return NutchApp.jobMgr.stop(cid, jid); - } else if (cmd.equals(Params.JOB_CMD_ABORT)) { - return NutchApp.jobMgr.abort(cid, jid); - } else if (cmd.equals(Params.JOB_CMD_GET)) { - return NutchApp.jobMgr.get(cid, jid); - } else { - throw new Exception("Unknown command: " + cmd); - } - } - } - - /* - * String crawlId - * String type - * String confId - * Object[] args - */ - @Put("json") - @SuppressWarnings("unchecked") - public Object create(Map args) throws Exception { - String cid = (String)args.get(Params.CRAWL_ID); - String typeString = (String)args.get(Params.JOB_TYPE); - JobType type = JobType.valueOf(typeString.toUpperCase()); - String confId = (String)args.get(Params.CONF_ID); - Object map = args.get(Params.ARGS); - Map cmdArgs = null; - if(map instanceof Map) - cmdArgs = (Map)map; - - String jobId = NutchApp.jobMgr.create(cid, type, confId, cmdArgs); - return jobId; - } -} diff --git src/java/org/apache/nutch/api/JobStatus.java src/java/org/apache/nutch/api/JobStatus.java deleted file mode 100644 index 61e2393..0000000 --- src/java/org/apache/nutch/api/JobStatus.java +++ /dev/null @@ -1,46 +0,0 @@ -/******************************************************************************* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - ******************************************************************************/ -package org.apache.nutch.api; - -import java.util.Map; - -import org.apache.nutch.api.JobManager.JobType; -import org.apache.nutch.util.NutchTool; - -public class JobStatus { - public static enum State {IDLE, RUNNING, FINISHED, FAILED, KILLED, - STOPPING, KILLING, ANY}; - public String id; - public JobType type; - public String confId; - public Map args; - public Map result; - public NutchTool tool; - public State state; - public String msg; - - public JobStatus(String id, JobType type, String confId, Map args, - State state, String msg) { - this.id = id; - this.type = type; - this.confId = confId; - this.args = args; - this.state = state; - this.msg = msg; - } - -} diff --git src/java/org/apache/nutch/api/NutchApp.java src/java/org/apache/nutch/api/NutchApp.java deleted file mode 100644 index 21c30ee..0000000 --- src/java/org/apache/nutch/api/NutchApp.java +++ /dev/null @@ -1,68 +0,0 @@ -/******************************************************************************* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - ******************************************************************************/ -package org.apache.nutch.api; - -import java.util.logging.Level; - -import org.apache.nutch.api.impl.RAMConfManager; -import org.apache.nutch.api.impl.RAMJobManager; -import org.restlet.Application; -import org.restlet.Restlet; -import org.restlet.routing.Router; - -public class NutchApp extends Application { - public static ConfManager confMgr; - public static JobManager jobMgr; - public static NutchServer server; - public static long started; - - static { - confMgr = new RAMConfManager(); - jobMgr = new RAMJobManager(); - } - - /** - * Creates a root Restlet that will receive all incoming calls. - */ - @Override - public synchronized Restlet createInboundRoot() { - getTunnelService().setEnabled(true); - getTunnelService().setExtensionsTunnel(true); - Router router = new Router(getContext()); - router.getLogger().setLevel(Level.FINEST); - // configs - router.attach("/", APIInfoResource.class); - router.attach("/" + AdminResource.PATH, AdminResource.class); - router.attach("/" + AdminResource.PATH + "/{" + Params.CMD + - "}", AdminResource.class); - router.attach("/" + ConfResource.PATH, ConfResource.class); - router.attach("/" + ConfResource.PATH + "/{"+ Params.CONF_ID + - "}", ConfResource.class); - router.attach("/" + ConfResource.PATH + "/{" + Params.CONF_ID + - "}/{" + Params.PROP_NAME + "}", ConfResource.class); - // db - router.attach("/" + DbResource.PATH, DbResource.class); - // jobs - router.attach("/" + JobResource.PATH, JobResource.class); - router.attach("/" + JobResource.PATH + "/{" + Params.JOB_ID + "}", - JobResource.class); - router.attach("/" + JobResource.PATH, JobResource.class); - router.attach("/" + JobResource.PATH + "/{" + Params.JOB_ID + "}/{" + - Params.CMD + "}", JobResource.class); - return router; - } -} diff --git src/java/org/apache/nutch/api/NutchServer.java src/java/org/apache/nutch/api/NutchServer.java index ab329ed..da53b7a 100644 --- src/java/org/apache/nutch/api/NutchServer.java +++ src/java/org/apache/nutch/api/NutchServer.java @@ -16,9 +16,14 @@ ******************************************************************************/ package org.apache.nutch.api; -import java.util.List; +import java.util.Collection; +import java.util.Set; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.TimeUnit; import java.util.logging.Level; +import javax.ws.rs.core.Application; + import org.apache.commons.cli.CommandLine; import org.apache.commons.cli.CommandLineParser; import org.apache.commons.cli.HelpFormatter; @@ -26,33 +31,52 @@ import org.apache.commons.cli.OptionBuilder; import org.apache.commons.cli.Options; import org.apache.commons.cli.PosixParser; import org.apache.commons.lang.StringUtils; -import org.apache.nutch.api.JobStatus.State; +import org.apache.nutch.api.impl.JobFactory; +import org.apache.nutch.api.impl.NutchServerPoolExecutor; +import org.apache.nutch.api.impl.RAMConfManager; +import org.apache.nutch.api.impl.RAMJobManager; +import org.apache.nutch.api.misc.ErrorStatusService; +import org.apache.nutch.api.model.response.JobInfo; +import org.apache.nutch.api.model.response.JobInfo.State; +import org.apache.nutch.api.resources.AdminResource; +import org.apache.nutch.api.resources.ConfigResource; +import org.apache.nutch.api.resources.DbResource; +import org.apache.nutch.api.resources.JobResource; import org.restlet.Component; +import org.restlet.Context; import org.restlet.data.Protocol; import org.restlet.data.Reference; -import org.restlet.representation.Representation; +import org.restlet.ext.jaxrs.JaxRsApplication; import org.restlet.resource.ClientResource; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -public class NutchServer { +import com.google.common.collect.Queues; +import com.google.common.collect.Sets; + +public class NutchServer extends Application { + public static final String NUTCH_SERVER = "NUTCH_SERVER"; + private static final Logger LOG = LoggerFactory.getLogger(NutchServer.class); private static final String LOCALHOST = "localhost"; private static final String DEFAULT_LOG_LEVEL = "INFO"; private static final Integer DEFAULT_PORT = 8081; - + private static final int JOB_CAPACITY = 100; + private static String logLevel = DEFAULT_LOG_LEVEL; private static Integer port = DEFAULT_PORT; - + private static final String CMD_HELP = "help"; private static final String CMD_STOP = "stop"; private static final String CMD_PORT = "port"; private static final String CMD_LOG_LEVEL = "log"; - private Component component; - private NutchApp app; + private ConfManager configManager; + private JobManager jobMgr; + private long started; + private boolean running; /** @@ -63,6 +87,13 @@ public class NutchServer { * level. */ public NutchServer() { + configManager = new RAMConfManager(); + BlockingQueue runnables = Queues + .newArrayBlockingQueue(JOB_CAPACITY); + NutchServerPoolExecutor executor = new NutchServerPoolExecutor(10, + JOB_CAPACITY, 1, TimeUnit.HOURS, runnables); + jobMgr = new RAMJobManager(new JobFactory(), executor, configManager); + // Create a new Component. component = new Component(); component.getLogger().setLevel(Level.parse(logLevel)); @@ -70,11 +101,36 @@ public class NutchServer { // Add a new HTTP server listening on defined port. component.getServers().add(Protocol.HTTP, port); + Context childContext = component.getContext().createChildContext(); + JaxRsApplication application = new JaxRsApplication(childContext); + application.add(this); + application.setStatusService(new ErrorStatusService()); + childContext.getAttributes().put(NUTCH_SERVER, this); + // Attach the application. - app = new NutchApp(); - component.getDefaultHost().attach("/nutch", app); + component.getDefaultHost().attach(application); + } + + @Override + public Set> getClasses() { + Set> resources = Sets.newHashSet(); + resources.add(JobResource.class); + resources.add(AdminResource.class); + resources.add(ConfigResource.class); + resources.add(DbResource.class); + return resources; + } + + public ConfManager getConfMgr() { + return configManager; + } + + public JobManager getJobMgr() { + return jobMgr; + } - NutchApp.server = this; + public long getStarted() { + return started; } /** @@ -89,29 +145,38 @@ public class NutchServer { /** * Starts the Nutch server printing some logging to the log file. * - * @throws Exception */ - public void start() throws Exception { + public void start() { LOG.info("Starting NutchServer on port: {} with logging level: {} ...", port, logLevel); - component.start(); + try { + component.start(); + } catch (Exception e) { + throw new IllegalStateException("Cannot start server!", e); + } LOG.info("Started NutchServer on port {}", port); running = true; - NutchApp.started = System.currentTimeMillis(); + started = System.currentTimeMillis(); } /** * Safety and convenience method to determine whether or not it is safe to * shut down the server. We make this assertion by consulting the - * {@link org.apache.nutch.api.NutchApp#jobMgr} for a list of jobs with - * {@link org.apache.nutch.api.JobStatus#state} equal to 'RUNNING'. + * {@link org.apache.nutch.api.NutchApp#jobManager} for a list of jobs with + * {@link org.apache.nutch.api.model.response.JobInfo#state} equal to 'RUNNING'. + * + * @param force + * ignore running tasks * * @return true if there are no jobs running or false if there are jobs with * running state. - * @throws Exception */ - public boolean canStop() throws Exception { - List jobs = NutchApp.jobMgr.list(null, State.RUNNING); + public boolean canStop(boolean force) { + if (force) { + return true; + } + + Collection jobs = getJobMgr().list(null, State.RUNNING); return jobs.isEmpty(); } @@ -123,18 +188,22 @@ public class NutchServer { * @return true if no server is running or if the shutdown was successful. * Return false if there are running jobs and the force switch has not * been activated. - * @throws Exception */ - public boolean stop(boolean force) throws Exception { - if (!NutchApp.server.running) { + public boolean stop(boolean force) { + if (!running) { return true; } - if (!NutchApp.server.canStop() && !force) { + if (!canStop(force)) { LOG.warn("Running jobs - can't stop now."); return false; } + LOG.info("Stopping NutchServer on port {}...", port); - component.stop(); + try { + component.stop(); + } catch (Exception e) { + throw new IllegalStateException("Cannot stop nutch server", e); + } LOG.info("Stopped NutchServer on port {}", port); running = false; return true; @@ -150,53 +219,54 @@ public class NutchServer { formatter.printHelp("NutchServer", options, true); return; } - + if (commandLine.hasOption(CMD_LOG_LEVEL)) { logLevel = commandLine.getOptionValue(CMD_LOG_LEVEL); } - + if (commandLine.hasOption(CMD_PORT)) { port = Integer.parseInt(commandLine.getOptionValue(CMD_PORT)); } if (commandLine.hasOption(CMD_STOP)) { String stopParameter = commandLine.getOptionValue(CMD_STOP); - boolean force = StringUtils.equals(Params.FORCE, stopParameter); + boolean force = StringUtils.equals("force", stopParameter); stopRemoteServer(force); return; } - + startServer(); } - - private static void startServer() throws Exception { + + private static void startServer() { NutchServer server = new NutchServer(); server.start(); } - - private static void stopRemoteServer(boolean force) throws Exception { + + private static void stopRemoteServer(boolean force) { Reference reference = new Reference(Protocol.HTTP, LOCALHOST, port); - reference.setPath("/nutch/admin/stop"); - + reference.setPath("/admin/stop"); + if (force) { - reference.addQueryParameter(Params.FORCE, Params.TRUE); + reference.addQueryParameter("force", "true"); } - + ClientResource clientResource = new ClientResource(reference); - Representation response = clientResource.get(); - LOG.info("Server response: {} ", response.getText()); + clientResource.get(); } private static Options createOptions() { Options options = new Options(); OptionBuilder.hasArg(); OptionBuilder.withArgName("logging level"); - OptionBuilder.withDescription("Select a logging level for the NutchServer: \n" - + "ALL|CONFIG|FINER|FINEST|INFO|OFF|SEVERE|WARNING"); + OptionBuilder + .withDescription("Select a logging level for the NutchServer: \n" + + "ALL|CONFIG|FINER|FINEST|INFO|OFF|SEVERE|WARNING"); options.addOption(OptionBuilder.create(CMD_LOG_LEVEL)); - OptionBuilder.withDescription("Stop running NutchServer. " - + "true value forces the Server to stop despite running jobs e.g. kills the tasks "); + OptionBuilder + .withDescription("Stop running NutchServer. " + + "true value forces the Server to stop despite running jobs e.g. kills the tasks "); OptionBuilder.hasOptionalArg(); OptionBuilder.withArgName("force"); options.addOption(OptionBuilder.create(CMD_STOP)); diff --git src/java/org/apache/nutch/api/Params.java src/java/org/apache/nutch/api/Params.java deleted file mode 100644 index 0b139e1..0000000 --- src/java/org/apache/nutch/api/Params.java +++ /dev/null @@ -1,38 +0,0 @@ -/******************************************************************************* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - ******************************************************************************/ -package org.apache.nutch.api; - -public interface Params { - - public static final String CONF_ID = "conf"; - public static final String PROP_NAME = "prop"; - public static final String PROP_VALUE = "value"; - public static final String PROPS = "props"; - public static final String CRAWL_ID = "crawl"; - public static final String JOB_ID = "job"; - public static final String JOB_TYPE = "type"; - public static final String ARGS = "args"; - public static final String CMD = "cmd"; - public static final String FORCE = "force"; - - - public static final String JOB_CMD_STOP = "stop"; - public static final String JOB_CMD_ABORT = "abort"; - public static final String JOB_CMD_GET = "get"; - - public static final String TRUE = "true"; -} diff --git src/java/org/apache/nutch/api/impl/JobFactory.java src/java/org/apache/nutch/api/impl/JobFactory.java new file mode 100644 index 0000000..d029a5b --- /dev/null +++ src/java/org/apache/nutch/api/impl/JobFactory.java @@ -0,0 +1,72 @@ +/******************************************************************************* + * 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.nutch.api.impl; + +import java.util.Map; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.util.ReflectionUtils; +import org.apache.nutch.api.JobManager.JobType; +import org.apache.nutch.crawl.DbUpdaterJob; +import org.apache.nutch.crawl.GeneratorJob; +import org.apache.nutch.crawl.InjectorJob; +import org.apache.nutch.crawl.WebTableReader; +import org.apache.nutch.fetcher.FetcherJob; +import org.apache.nutch.indexer.IndexingJob; +import org.apache.nutch.parse.ParserJob; +import org.apache.nutch.util.NutchTool; + +import com.google.common.collect.Maps; + +public class JobFactory { + private static Map> typeToClass; + + static { + typeToClass = Maps.newHashMap(); + typeToClass.put(JobType.FETCH, FetcherJob.class); + typeToClass.put(JobType.GENERATE, GeneratorJob.class); + typeToClass.put(JobType.INDEX, IndexingJob.class); + typeToClass.put(JobType.INJECT, InjectorJob.class); + typeToClass.put(JobType.PARSE, ParserJob.class); + typeToClass.put(JobType.UPDATEDB, DbUpdaterJob.class); + typeToClass.put(JobType.READDB, WebTableReader.class); + } + + public NutchTool createToolByType(JobType type, Configuration conf) { + if (!typeToClass.containsKey(type)) { + return null; + } + Class clz = typeToClass.get(type); + return createTool(clz, conf); + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + public NutchTool createToolByClassName(String className, Configuration conf) { + try { + Class clz = Class.forName(className); + return createTool(clz, conf); + } catch (ClassNotFoundException e) { + throw new IllegalStateException(e); + } + } + + private NutchTool createTool(Class clz, + Configuration conf) { + return ReflectionUtils.newInstance(clz, conf); + } + +} diff --git src/java/org/apache/nutch/api/impl/JobWorker.java src/java/org/apache/nutch/api/impl/JobWorker.java new file mode 100644 index 0000000..9c7c5c2 --- /dev/null +++ src/java/org/apache/nutch/api/impl/JobWorker.java @@ -0,0 +1,103 @@ +/******************************************************************************* + * 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.nutch.api.impl; + +import java.text.MessageFormat; + +import org.apache.hadoop.conf.Configuration; +import org.apache.nutch.api.model.request.JobConfig; +import org.apache.nutch.api.model.response.JobInfo; +import org.apache.nutch.api.model.response.JobInfo.State; +import org.apache.nutch.api.resources.ConfigResource; +import org.apache.nutch.metadata.Nutch; +import org.apache.nutch.util.NutchTool; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class JobWorker implements Runnable { + private static final Logger LOG = LoggerFactory.getLogger(JobWorker.class); + private NutchTool tool; + private JobInfo jobInfo; + private JobConfig jobConfig; + + public JobWorker(JobConfig jobConfig, Configuration conf, NutchTool tool) { + this.tool = tool; + this.jobConfig = jobConfig; + if (jobConfig.getConfId() == null) { + jobConfig.setConfId(ConfigResource.DEFAULT); + } + + jobInfo = new JobInfo(generateId(), jobConfig, State.IDLE, "idle"); + if (jobConfig.getCrawlId() != null) { + conf.set(Nutch.CRAWL_ID_KEY, jobConfig.getCrawlId()); + } + } + + private String generateId() { + if (jobConfig.getCrawlId() == null) { + return MessageFormat.format("{0}-{1}-{2}", jobConfig.getConfId(), + jobConfig.getType(), String.valueOf(hashCode())); + } + return MessageFormat.format("{0}-{1}-{2}-{3}", jobConfig.getCrawlId(), + jobConfig.getConfId(), jobConfig.getType(), String.valueOf(hashCode())); + } + + @Override + public void run() { + try { + getInfo().setState(State.RUNNING); + getInfo().setMsg("OK"); + getInfo().setResult(tool.run(getInfo().getArgs())); + getInfo().setState(State.FINISHED); + } catch (Exception e) { + LOG.error("Cannot run job worker!", e); + getInfo().setMsg("ERROR: " + e.toString()); + getInfo().setState(State.FAILED); + } + } + + public boolean stopJob() { + getInfo().setState(State.STOPPING); + try { + return tool.stopJob(); + } catch (Exception e) { + throw new RuntimeException( + "Cannot stop job with id " + getInfo().getId(), e); + } + } + + public boolean killJob() { + getInfo().setState(State.KILLING); + try { + boolean result = tool.killJob(); + getInfo().setState(State.KILLED); + return result; + } catch (Exception e) { + throw new RuntimeException( + "Cannot kill job with id " + getInfo().getId(), e); + } + } + + public JobInfo getInfo() { + return jobInfo; + } + + public void setInfo(JobInfo jobInfo) { + this.jobInfo = jobInfo; + } + +} \ No newline at end of file diff --git src/java/org/apache/nutch/api/impl/NutchServerPoolExecutor.java src/java/org/apache/nutch/api/impl/NutchServerPoolExecutor.java new file mode 100644 index 0000000..fa4c58d --- /dev/null +++ src/java/org/apache/nutch/api/impl/NutchServerPoolExecutor.java @@ -0,0 +1,103 @@ +/******************************************************************************* + * 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.nutch.api.impl; + +import java.util.Collection; +import java.util.List; +import java.util.Queue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +import org.apache.commons.collections.CollectionUtils; +import org.apache.commons.lang.StringUtils; +import org.apache.nutch.api.model.response.JobInfo; + +import com.google.common.collect.Lists; +import com.google.common.collect.Queues; + +public class NutchServerPoolExecutor extends ThreadPoolExecutor { + + private Queue workersHistory; + private Queue runningWorkers; + + public NutchServerPoolExecutor(int corePoolSize, int maximumPoolSize, + long keepAliveTime, TimeUnit unit, BlockingQueue workQueue) { + super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue); + workersHistory = Queues.newArrayBlockingQueue(maximumPoolSize); + runningWorkers = Queues.newArrayBlockingQueue(maximumPoolSize); + } + + @Override + protected void beforeExecute(Thread thread, Runnable runnable) { + super.beforeExecute(thread, runnable); + synchronized (runningWorkers) { + runningWorkers.offer(((JobWorker) runnable)); + } + } + + @Override + protected void afterExecute(Runnable runnable, Throwable throwable) { + super.afterExecute(runnable, throwable); + synchronized (runningWorkers) { + runningWorkers.remove(((JobWorker) runnable).getInfo()); + } + JobWorker worker = ((JobWorker) runnable); + addStatusToHistory(worker); + } + + private void addStatusToHistory(JobWorker worker) { + synchronized (workersHistory) { + if (!workersHistory.offer(worker)) { + workersHistory.poll(); + workersHistory.add(worker); + } + } + } + + public JobWorker findWorker(String jobId) { + synchronized (runningWorkers) { + for (JobWorker worker : runningWorkers) { + if (StringUtils.equals(worker.getInfo().getId(), jobId)) { + return worker; + } + } + } + return null; + } + + public Collection getJobHistory() { + return getJobsInfo(workersHistory); + } + + public Collection getJobRunning() { + return getJobsInfo(runningWorkers); + } + + @SuppressWarnings("unchecked") + public Collection getAllJobs() { + return CollectionUtils.union(getJobRunning(), getJobHistory()); + } + + private Collection getJobsInfo(Collection workers) { + List jobsInfo = Lists.newLinkedList(); + for (JobWorker worker : workers) { + jobsInfo.add(worker.getInfo()); + } + return jobsInfo; + } +} diff --git src/java/org/apache/nutch/api/impl/RAMConfManager.java src/java/org/apache/nutch/api/impl/RAMConfManager.java index 59c7412..e7d8496 100644 --- src/java/org/apache/nutch/api/impl/RAMConfManager.java +++ src/java/org/apache/nutch/api/impl/RAMConfManager.java @@ -16,68 +16,104 @@ ******************************************************************************/ package org.apache.nutch.api.impl; -import java.util.HashMap; +import java.util.Collections; import java.util.Iterator; import java.util.Map; -import java.util.Set; import java.util.Map.Entry; -import java.util.TreeMap; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.commons.collections.MapUtils; +import org.apache.commons.lang.StringUtils; import org.apache.hadoop.conf.Configuration; import org.apache.nutch.api.ConfManager; -import org.apache.nutch.api.ConfResource; +import org.apache.nutch.api.model.request.NutchConfig; +import org.apache.nutch.api.resources.ConfigResource; import org.apache.nutch.util.NutchConfiguration; +import com.google.common.collect.Maps; + public class RAMConfManager implements ConfManager { - Map configs = new HashMap(); - + private Map configurations = Maps.newConcurrentMap(); + + private AtomicInteger newConfigId = new AtomicInteger(); + public RAMConfManager() { - configs.put(ConfResource.DEFAULT_CONF, NutchConfiguration.create()); + configurations.put(ConfigResource.DEFAULT, NutchConfiguration.create()); } - + public Set list() { - return configs.keySet(); + return configurations.keySet(); } - + public Configuration get(String confId) { - return configs.get(confId); - } - - public Map getAsMap(String confId) { - Configuration cfg = configs.get(confId); - if (cfg == null) return null; - Iterator> it = cfg.iterator(); - TreeMap res = new TreeMap(); - while (it.hasNext()) { - Entry e = it.next(); - res.put(e.getKey(), e.getValue()); + if (confId == null) { + return configurations.get(ConfigResource.DEFAULT); } - return res; + return configurations.get(confId); } - - public void create(String confId, Map props, boolean force) throws Exception { - if (configs.containsKey(confId) && !force) { - throw new Exception("Config name '" + confId + "' already exists."); + + public Map getAsMap(String confId) { + Configuration configuration = configurations.get(confId); + if (configuration == null) { + return Collections.emptyMap(); } - Configuration conf = NutchConfiguration.create(); - // apply overrides - if (props != null) { - for (Entry e : props.entrySet()) { - conf.set(e.getKey(), e.getValue()); - } + + Iterator> iterator = configuration.iterator(); + Map configMap = Maps.newTreeMap(); + while (iterator.hasNext()) { + Entry entry = iterator.next(); + configMap.put(entry.getKey(), entry.getValue()); } - configs.put(confId, conf); + return configMap; } - - public void setProperty(String confId, String propName, String propValue) throws Exception { - if (!configs.containsKey(confId)) { - throw new Exception("Unknown configId '" + confId + "'"); + + public void setProperty(String confId, String propName, String propValue) { + if (!configurations.containsKey(confId)) { + throw new IllegalArgumentException("Unknown configId '" + confId + "'"); } - Configuration conf = configs.get(confId); + Configuration conf = configurations.get(confId); conf.set(propName, propValue); } - + public void delete(String confId) { - configs.remove(confId); + configurations.remove(confId); } + + @Override + public String create(NutchConfig nutchConfig) { + if (StringUtils.isBlank(nutchConfig.getConfigId())) { + nutchConfig.setConfigId(String.valueOf(newConfigId.incrementAndGet())); + } + + if (!canCreate(nutchConfig)) { + throw new IllegalArgumentException("Config already exists."); + } + + createHadoopConfig(nutchConfig); + return nutchConfig.getConfigId(); + } + + private boolean canCreate(NutchConfig nutchConfig) { + if (nutchConfig.isForce()) { + return true; + } + if (!configurations.containsKey(nutchConfig.getConfigId())) { + return true; + } + return false; + } + + private void createHadoopConfig(NutchConfig nutchConfig) { + Configuration conf = NutchConfiguration.create(); + configurations.put(nutchConfig.getConfigId(), conf); + + if (MapUtils.isEmpty(nutchConfig.getParams())) { + return; + } + for (Entry e : nutchConfig.getParams().entrySet()) { + conf.set(e.getKey(), e.getValue()); + } + } + } diff --git src/java/org/apache/nutch/api/impl/RAMJobManager.java src/java/org/apache/nutch/api/impl/RAMJobManager.java index 226804f..1376093 100644 --- src/java/org/apache/nutch/api/impl/RAMJobManager.java +++ src/java/org/apache/nutch/api/impl/RAMJobManager.java @@ -16,237 +16,83 @@ ******************************************************************************/ package org.apache.nutch.api.impl; -import java.util.ArrayList; -import java.util.Collections; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ArrayBlockingQueue; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.ThreadPoolExecutor; -import java.util.concurrent.TimeUnit; +import java.util.Collection; +import org.apache.commons.lang.StringUtils; import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.util.ReflectionUtils; -import org.apache.nutch.api.ConfResource; +import org.apache.nutch.api.ConfManager; import org.apache.nutch.api.JobManager; -import org.apache.nutch.api.JobStatus; -import org.apache.nutch.api.JobStatus.State; -import org.apache.nutch.api.NutchApp; -import org.apache.nutch.crawl.DbUpdaterJob; -import org.apache.nutch.crawl.GeneratorJob; -import org.apache.nutch.crawl.InjectorJob; -import org.apache.nutch.crawl.WebTableReader; -import org.apache.nutch.fetcher.FetcherJob; -import org.apache.nutch.indexer.IndexingJob; -import org.apache.nutch.metadata.Nutch; -import org.apache.nutch.parse.ParserJob; +import org.apache.nutch.api.model.request.JobConfig; +import org.apache.nutch.api.model.response.JobInfo; +import org.apache.nutch.api.model.response.JobInfo.State; import org.apache.nutch.util.NutchTool; public class RAMJobManager implements JobManager { - int CAPACITY = 100; - ThreadPoolExecutor exec = new MyPoolExecutor(10, CAPACITY, 1, TimeUnit.HOURS, - new ArrayBlockingQueue(CAPACITY)); - - private class MyPoolExecutor extends ThreadPoolExecutor { + private JobFactory jobFactory; + private NutchServerPoolExecutor executor; + private ConfManager configManager; - public MyPoolExecutor(int corePoolSize, int maximumPoolSize, - long keepAliveTime, TimeUnit unit, BlockingQueue workQueue) { - super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue); - } + public RAMJobManager(JobFactory jobFactory, NutchServerPoolExecutor executor, + ConfManager configManager) { + this.jobFactory = jobFactory; + this.executor = executor; + this.configManager = configManager; + } - @Override - protected void beforeExecute(Thread t, Runnable r) { - // TODO Auto-generated method stub - super.beforeExecute(t, r); - synchronized(jobRunning) { - jobRunning.offer(((JobWorker)r).jobStatus); - } + @Override + public Collection list(String crawlId, State state) { + if (state == null || state == State.ANY) { + return executor.getAllJobs(); } - - @Override - protected void afterExecute(Runnable r, Throwable t) { - super.afterExecute(r, t); - synchronized(jobRunning) { - jobRunning.remove(((JobWorker)r).jobStatus); - } - JobStatus status = ((JobWorker)r).jobStatus; - synchronized(jobHistory) { - if (!jobHistory.offer(status)) { - jobHistory.poll(); - jobHistory.add(status); - } - } + if (state == State.RUNNING || state == State.IDLE) { + return executor.getJobRunning(); } - } - - ArrayBlockingQueue jobHistory = new ArrayBlockingQueue(CAPACITY); - ArrayBlockingQueue jobRunning = new ArrayBlockingQueue(CAPACITY); - - private static Map> typeToClass = new HashMap>(); - - static { - typeToClass.put(JobType.FETCH, FetcherJob.class); - typeToClass.put(JobType.GENERATE, GeneratorJob.class); - typeToClass.put(JobType.INDEX, IndexingJob.class); - typeToClass.put(JobType.INJECT, InjectorJob.class); - typeToClass.put(JobType.PARSE, ParserJob.class); - typeToClass.put(JobType.UPDATEDB, DbUpdaterJob.class); - typeToClass.put(JobType.READDB, WebTableReader.class); + return executor.getJobHistory(); } - private void addFinishedStatus(JobStatus status) { - synchronized(jobHistory) { - if (!jobHistory.offer(status)) { - jobHistory.poll(); - jobHistory.add(status); - } - } - } - @Override - @SuppressWarnings("fallthrough") - public List list(String crawlId, State state) throws Exception { - List res = new ArrayList(); - if (state == null) state = State.ANY; - switch(state) { - case ANY: - res.addAll(jobHistory); - /* FALLTHROUGH */ - case RUNNING: - case IDLE: - res.addAll(jobRunning); - break; - default: - res.addAll(jobHistory); - } - return res; + public JobInfo get(String crawlId, String jobId) { + return executor.findWorker(jobId).getInfo(); } @Override - public JobStatus get(String crawlId, String jobId) throws Exception { - for (JobStatus job : jobRunning) { - if (job.id.equals(jobId)) { - return job; - } + public String create(JobConfig jobConfig) { + if (jobConfig.getArgs() == null) { + throw new IllegalArgumentException("Arguments cannot be null!"); } - for (JobStatus job : jobHistory) { - if (job.id.equals(jobId)) { - return job; - } - } - return null; + + Configuration conf = cloneConfiguration(jobConfig.getConfId()); + NutchTool tool = createTool(jobConfig, conf); + JobWorker worker = new JobWorker(jobConfig, conf, tool); + + executor.execute(worker); + executor.purge(); + return worker.getInfo().getId(); } - @Override - public String create(String crawlId, JobType type, String confId, - Map args) throws Exception { - if (args == null) args = Collections.emptyMap(); - JobWorker worker = new JobWorker(crawlId, type, confId, args); - String id = worker.getId(); - exec.execute(worker); - exec.purge(); - return id; + private Configuration cloneConfiguration(String confId) { + Configuration conf = configManager.get(confId); + if (conf == null) { + throw new IllegalArgumentException("Unknown confId " + confId); + } + return new Configuration(conf); } - @Override - public boolean abort(String crawlId, String id) throws Exception { - // find running job - for (JobStatus job : jobRunning) { - if (job.id.equals(id)) { - job.state = State.KILLING; - boolean res = job.tool.killJob(); - job.state = State.KILLED; - return res; - } + private NutchTool createTool(JobConfig jobConfig, Configuration conf) { + if (StringUtils.isNotBlank(jobConfig.getJobClassName())) { + return jobFactory.createToolByClassName(jobConfig.getJobClassName(), conf); } - return false; + + return jobFactory.createToolByType(jobConfig.getType(), conf); } @Override - public boolean stop(String crawlId, String id) throws Exception { - // find running job - for (JobStatus job : jobRunning) { - if (job.id.equals(id)) { - job.state = State.STOPPING; - boolean res = job.tool.stopJob(); - return res; - } - } - return false; + public boolean abort(String crawlId, String id) { + return executor.findWorker(id).killJob(); } - - private class JobWorker implements Runnable { - String id; - JobType type; - String confId; - NutchTool tool; - Map args; - JobStatus jobStatus; - - @SuppressWarnings("unchecked") - JobWorker(String crawlId, JobType type, String confId, Map args) throws Exception { - if (confId == null) { - confId = ConfResource.DEFAULT_CONF; - } - Configuration conf = NutchApp.confMgr.get(confId); - // clone it - we are going to modify it - if (conf == null) { - throw new Exception("Unknown confId " + confId); - } - this.id = confId + "-" + type + "-" + hashCode(); - this.type = type; - this.confId = confId; - this.args = args; - conf = new Configuration(conf); - if (crawlId != null) { - conf.set(Nutch.CRAWL_ID_KEY, crawlId); - this.id = crawlId + "-" + this.id; - } - Class clz = typeToClass.get(type); - if (clz == null) { - Class c = Class.forName((String)args.get(Nutch.ARG_CLASS)); - if(c instanceof Class) { - clz = (Class) c; - } - } - tool = ReflectionUtils.newInstance(clz, conf); - jobStatus = new JobStatus(id, type, confId, args, State.IDLE, "idle"); - jobStatus.tool = tool; - } - - public String getId() { - return id; - } - - public float getProgress() { - return tool.getProgress(); - } - - public State getState() { - return jobStatus.state; - } - - public Map getResult() { - return jobStatus.result; - } - - public Map getStatus() { - return tool.getStatus(); - } - @Override - public void run() { - try { - jobStatus.state = State.RUNNING; - jobStatus.msg = "OK"; - jobStatus.result = tool.run(args); - jobStatus.state = State.FINISHED; - } catch (Exception e) { - e.printStackTrace(); - jobStatus.msg = "ERROR: " + e.toString(); - jobStatus.state = State.FAILED; - } - } + @Override + public boolean stop(String crawlId, String id) { + return executor.findWorker(id).stopJob(); } } diff --git src/java/org/apache/nutch/api/impl/db/DbIterator.java src/java/org/apache/nutch/api/impl/db/DbIterator.java new file mode 100644 index 0000000..5a6770e --- /dev/null +++ src/java/org/apache/nutch/api/impl/db/DbIterator.java @@ -0,0 +1,111 @@ +/******************************************************************************* + * 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.nutch.api.impl.db; + +import java.io.IOException; +import java.util.Map; +import java.util.Set; + +import org.apache.avro.util.Utf8; +import org.apache.commons.collections.CollectionUtils; +import org.apache.gora.query.Result; +import org.apache.nutch.storage.Mark; +import org.apache.nutch.storage.WebPage; +import org.apache.nutch.util.NutchJob; +import org.apache.nutch.util.TableUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.collect.Sets; +import com.google.common.collect.UnmodifiableIterator; + +public class DbIterator extends UnmodifiableIterator> { + private static final Logger LOG = LoggerFactory.getLogger(DbIterator.class); + + private Result result; + private boolean hasNext; + private String url; + private WebPage page; + private Utf8 batchId; + private Set commonFields; + + DbIterator(Result res, Set fields, String batchId) { + this.result = res; + if (batchId != null) { + this.batchId = new Utf8(batchId); + } + if (fields != null) { + this.commonFields = Sets.newTreeSet(fields); + } + try { + skipNonRelevant(); + } catch (Exception e) { + LOG.error("Cannot create db iterator!", e); + } + } + + private void skipNonRelevant() throws Exception, IOException { + hasNext = result.next(); + if (!hasNext) { + return; + } + if (batchId == null) { + return; + } + + while (hasNext) { + WebPage page = result.get(); + Utf8 mark = Mark.UPDATEDB_MARK.checkMark(page); + if (NutchJob.shouldProcess(mark, batchId)) { + return; + } + + LOG.debug("Skipping {}; different batch id", result.getKey()); + hasNext = result.next(); + } + } + + public boolean hasNext() { + return hasNext; + } + + public Map next() { + url = result.getKey(); + page = WebPage.newBuilder(result.get()).build(); + try { + skipNonRelevant(); + if (!hasNext) { + result.close(); + } + } catch (Exception e) { + LOG.error("Cannot get next result!", e); + hasNext = false; + return null; + } + return pageAsMap(url, page); + } + + private Map pageAsMap(String url, WebPage page) { + Map result = DbPageConverter.convertPage(page, commonFields); + + if (CollectionUtils.isEmpty(commonFields) || commonFields.contains("url")) { + result.put("url", TableUtil.unreverseUrl(url)); + } + return result; + } + +} \ No newline at end of file diff --git src/java/org/apache/nutch/api/impl/db/DbPageConverter.java src/java/org/apache/nutch/api/impl/db/DbPageConverter.java new file mode 100644 index 0000000..b0ae267 --- /dev/null +++ src/java/org/apache/nutch/api/impl/db/DbPageConverter.java @@ -0,0 +1,135 @@ +/******************************************************************************* + * 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.nutch.api.impl.db; + +import java.nio.ByteBuffer; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Set; + +import org.apache.avro.Schema.Field; +import org.apache.avro.util.Utf8; +import org.apache.commons.collections.CollectionUtils; +import org.apache.commons.collections.MapUtils; +import org.apache.commons.lang.StringUtils; +import org.apache.nutch.parse.ParseStatusUtils; +import org.apache.nutch.protocol.ProtocolStatusUtils; +import org.apache.nutch.storage.WebPage; +import org.apache.nutch.util.Bytes; +import org.apache.nutch.util.StringUtil; + +import com.google.common.collect.Maps; +import com.google.common.collect.Sets; + +public class DbPageConverter { + + public static Map convertPage(WebPage page, Set fields) { + Map result = Maps.newHashMap(); + for (Field field : filterFields(page, fields)) { + Object value = convertField(page, field); + if (value != null) { + result.put(field.name(), value); + } + } + return result; + } + + private static Object convertField(WebPage page, Field field) { + int index = field.pos(); + if (index < 0) { + return null; + } + + Object value = page.get(index); + if (value == null) { + return null; + } + + String fieldName = field.name(); + if (StringUtils.equals(fieldName, "metadata")) { + return getSimpleMetadata(page); + } + if (StringUtils.equals(fieldName, "protocolStatus")) { + return ProtocolStatusUtils.toString(page.getProtocolStatus()); + } + if (StringUtils.equals(fieldName, "parseStatus")) { + return ParseStatusUtils.toString(page.getParseStatus()); + } + if (StringUtils.equals(fieldName, "signature")) { + return StringUtil.toHexString(page.getSignature()); + } + if (StringUtils.equals(fieldName, "content")) { + return Bytes.toStringBinary(page.getContent()); + } + if (StringUtils.equals(fieldName, "markers")) { + return convertToStringsMap(page.getMarkers()); + } + if (StringUtils.equals(fieldName, "inlinks")) { + return convertToStringsMap(page.getInlinks()); + } + if (StringUtils.equals(fieldName, "outlinks")) { + return convertToStringsMap(page.getOutlinks()); + } + + if (value instanceof Utf8) { + return value.toString(); + } + + if (value instanceof ByteBuffer) { + return Bytes.toStringBinary((ByteBuffer) value); + } + + return value; + } + + private static Set filterFields(WebPage page, Set queryFields) { + List pageFields = page.getSchema().getFields(); + if (CollectionUtils.isEmpty(queryFields)) { + return Sets.newHashSet(pageFields); + } + + Set filteredFields = Sets.newLinkedHashSet(); + for (Field field : pageFields) { + if (queryFields.contains(field.name())) { + filteredFields.add(field); + } + } + return filteredFields; + } + + private static Map getSimpleMetadata(WebPage page) { + Map metadata = page.getMetadata(); + if (MapUtils.isEmpty(metadata)) { + return Collections.emptyMap(); + } + Map simpleMeta = Maps.newHashMap(); + for (CharSequence key : metadata.keySet()) { + simpleMeta.put(key.toString(), Bytes.toStringBinary(metadata.get(key))); + } + return simpleMeta; + } + + private static Map convertToStringsMap(Map map) { + Map res = Maps.newHashMap(); + for (Entry entry : map.entrySet()) { + res.put(entry.getKey().toString(), entry.getValue().toString()); + } + return res; + } +} diff --git src/java/org/apache/nutch/api/impl/db/DbReader.java src/java/org/apache/nutch/api/impl/db/DbReader.java new file mode 100644 index 0000000..20084b7 --- /dev/null +++ src/java/org/apache/nutch/api/impl/db/DbReader.java @@ -0,0 +1,75 @@ +package org.apache.nutch.api.impl.db; + +import java.net.MalformedURLException; +import java.util.Iterator; +import java.util.Map; +import java.util.Set; + +import org.apache.commons.collections.CollectionUtils; +import org.apache.commons.lang.StringUtils; +import org.apache.gora.query.Query; +import org.apache.gora.query.Result; +import org.apache.gora.store.DataStore; +import org.apache.hadoop.conf.Configuration; +import org.apache.nutch.api.model.request.DbFilter; +import org.apache.nutch.metadata.Nutch; +import org.apache.nutch.storage.StorageUtils; +import org.apache.nutch.storage.WebPage; +import org.apache.nutch.util.TableUtil; + +public class DbReader { + private DataStore store; + + public DbReader(Configuration conf, String crawlId) { + conf = new Configuration(conf); + if (crawlId != null) { + conf.set(Nutch.CRAWL_ID_KEY, crawlId); + } + try { + store = StorageUtils.createWebStore(conf, String.class, WebPage.class); + } catch (Exception e) { + throw new IllegalStateException("Cannot create webstore!", e); + } + } + + public Iterator> runQuery(DbFilter filter) { + String startKey = filter.getStartKey(); + String endKey = filter.getEndKey(); + + if (!filter.isKeysReversed()) { + startKey = reverseKey(filter.getStartKey()); + endKey = reverseKey(filter.getEndKey()); + } + + Query query = store.newQuery(); + query.setFields(prepareFields(filter.getFields())); + if (startKey != null) { + query.setStartKey(startKey); + if (endKey != null) { + query.setEndKey(endKey); + } + } + Result result = store.execute(query); + return new DbIterator(result, filter.getFields(), filter.getBatchId()); + } + + private String reverseKey(String key) { + if (StringUtils.isEmpty(key)) { + return null; + } + + try { + return TableUtil.reverseUrl(key); + } catch (MalformedURLException e) { + throw new IllegalArgumentException("Wrong url format!", e); + } + } + + private String[] prepareFields(Set fields) { + if (CollectionUtils.isEmpty(fields)) { + return null; + } + fields.remove("url"); + return fields.toArray(new String[fields.size()]); + } +} diff --git src/java/org/apache/nutch/api/misc/ErrorStatusService.java src/java/org/apache/nutch/api/misc/ErrorStatusService.java new file mode 100644 index 0000000..0f0aa33 --- /dev/null +++ src/java/org/apache/nutch/api/misc/ErrorStatusService.java @@ -0,0 +1,40 @@ +/******************************************************************************* + * 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.nutch.api.misc; + +import org.apache.nutch.api.model.response.ErrorResponse; +import org.restlet.Request; +import org.restlet.Response; +import org.restlet.data.Status; +import org.restlet.ext.jackson.JacksonRepresentation; +import org.restlet.representation.Representation; +import org.restlet.service.StatusService; + +public class ErrorStatusService extends StatusService { + @Override + public Status getStatus(Throwable throwable, Request request, + Response response) { + return new Status(Status.SERVER_ERROR_INTERNAL, throwable); + } + + @Override + public Representation getRepresentation(Status status, Request request, + Response response) { + ErrorResponse errorResponse = new ErrorResponse(status.getThrowable()); + return new JacksonRepresentation(errorResponse); + } +} diff --git src/java/org/apache/nutch/api/model/request/DbFilter.java src/java/org/apache/nutch/api/model/request/DbFilter.java new file mode 100644 index 0000000..f6ff3ce --- /dev/null +++ src/java/org/apache/nutch/api/model/request/DbFilter.java @@ -0,0 +1,67 @@ +/******************************************************************************* + * 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.nutch.api.model.request; + +import java.util.Set; + +public class DbFilter { + private String batchId; + private String startKey; + private String endKey; + private boolean isKeysReversed = false; + private Set fields; + + public Set getFields() { + return fields; + } + + public void setFields(Set fields) { + this.fields = fields; + } + + public boolean isKeysReversed() { + return isKeysReversed; + } + + public void setKeysReversed(boolean isKeysReversed) { + this.isKeysReversed = isKeysReversed; + } + + public String getEndKey() { + return endKey; + } + + public void setEndKey(String endKey) { + this.endKey = endKey; + } + + public String getStartKey() { + return startKey; + } + + public void setStartKey(String startKey) { + this.startKey = startKey; + } + + public String getBatchId() { + return batchId; + } + + public void setBatchId(String batchId) { + this.batchId = batchId; + } +} diff --git src/java/org/apache/nutch/api/model/request/JobConfig.java src/java/org/apache/nutch/api/model/request/JobConfig.java new file mode 100644 index 0000000..88ac739 --- /dev/null +++ src/java/org/apache/nutch/api/model/request/JobConfig.java @@ -0,0 +1,69 @@ +/******************************************************************************* + * 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.nutch.api.model.request; + +import java.util.Map; + +import org.apache.nutch.api.JobManager.JobType; + +public class JobConfig { + private String crawlId; + private JobType type; + private String confId; + private String jobClassName; + private Map args; + + public String getCrawlId() { + return crawlId; + } + + public void setCrawlId(String crawlId) { + this.crawlId = crawlId; + } + + public JobType getType() { + return type; + } + + public void setType(JobType type) { + this.type = type; + } + + public String getConfId() { + return confId; + } + + public void setConfId(String confId) { + this.confId = confId; + } + + public Map getArgs() { + return args; + } + + public void setArgs(Map args) { + this.args = args; + } + + public String getJobClassName() { + return jobClassName; + } + + public void setJobClassName(String jobClass) { + this.jobClassName = jobClass; + } +} diff --git src/java/org/apache/nutch/api/model/request/NutchConfig.java src/java/org/apache/nutch/api/model/request/NutchConfig.java new file mode 100644 index 0000000..5596b66 --- /dev/null +++ src/java/org/apache/nutch/api/model/request/NutchConfig.java @@ -0,0 +1,51 @@ +/******************************************************************************* + * 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.nutch.api.model.request; + +import java.util.Map; + +import java.util.Collections; + +public class NutchConfig { + private String configId; + private boolean force = false; + private Map params = Collections.emptyMap(); + + public Map getParams() { + return params; + } + + public void setParams(Map params) { + this.params = params; + } + + public String getConfigId() { + return configId; + } + + public void setConfigId(String configId) { + this.configId = configId; + } + + public boolean isForce() { + return force; + } + + public void setForce(boolean force) { + this.force = force; + } +} diff --git src/java/org/apache/nutch/api/model/response/DbQueryResult.java src/java/org/apache/nutch/api/model/response/DbQueryResult.java new file mode 100644 index 0000000..3fc7a3a --- /dev/null +++ src/java/org/apache/nutch/api/model/response/DbQueryResult.java @@ -0,0 +1,35 @@ +/******************************************************************************* + * 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.nutch.api.model.response; + +import java.util.Collections; +import java.util.List; +import java.util.Map; + +import com.google.common.collect.Lists; + +public class DbQueryResult { + private List> values = Lists.newLinkedList(); + + public List> getValues() { + return Collections.unmodifiableList(values); + } + + public void addValue(Map next) { + values.add(next); + } +} diff --git src/java/org/apache/nutch/api/model/response/ErrorResponse.java src/java/org/apache/nutch/api/model/response/ErrorResponse.java new file mode 100644 index 0000000..ae1f6eb --- /dev/null +++ src/java/org/apache/nutch/api/model/response/ErrorResponse.java @@ -0,0 +1,47 @@ +/******************************************************************************* + * 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.nutch.api.model.response; + +import org.apache.commons.lang.exception.ExceptionUtils; + +public class ErrorResponse { + private String exception; + private String message; + private String stackTrace; + + public ErrorResponse(Throwable throwable) { + if (throwable == null) { + message = "Unknown error!"; + return; + } + exception = throwable.getClass().toString(); + message = ExceptionUtils.getMessage(throwable); + stackTrace = ExceptionUtils.getFullStackTrace(throwable); + } + + public String getException() { + return exception; + } + + public String getMessage() { + return message; + } + + public String getStackTrace() { + return stackTrace; + } +} diff --git src/java/org/apache/nutch/api/model/response/JobInfo.java src/java/org/apache/nutch/api/model/response/JobInfo.java new file mode 100644 index 0000000..5e7448e --- /dev/null +++ src/java/org/apache/nutch/api/model/response/JobInfo.java @@ -0,0 +1,113 @@ +/******************************************************************************* + * 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.nutch.api.model.response; + +import java.util.Map; + +import org.apache.nutch.api.JobManager.JobType; +import org.apache.nutch.api.model.request.JobConfig; + +public class JobInfo { + + public static enum State { + IDLE, RUNNING, FINISHED, FAILED, KILLED, STOPPING, KILLING, ANY + }; + + private String id; + private JobType type; + private String confId; + private Map args; + private Map result; + private State state; + private String msg; + private String crawlId; + + public JobInfo(String id, JobConfig config, State state, String msg) { + this.id = id; + this.crawlId = config.getCrawlId(); + this.type = config.getType(); + this.confId = config.getConfId(); + this.args = config.getArgs(); + this.state = state; + this.msg = msg; + } + + public String getMsg() { + return msg; + } + + public void setMsg(String msg) { + this.msg = msg; + } + + public State getState() { + return state; + } + + public void setState(State state) { + this.state = state; + } + + public Map getResult() { + return result; + } + + public void setResult(Map result) { + this.result = result; + } + + public Map getArgs() { + return args; + } + + public void setArgs(Map args) { + this.args = args; + } + + public String getConfId() { + return confId; + } + + public void setConfId(String confId) { + this.confId = confId; + } + + public JobType getType() { + return type; + } + + public void setType(JobType type) { + this.type = type; + } + + public String getId() { + return id; + } + + public void setId(String id) { + this.id = id; + } + + public String getCrawlId() { + return crawlId; + } + + public void setCrawlId(String crawlId) { + this.crawlId = crawlId; + } + +} diff --git src/java/org/apache/nutch/api/model/response/NutchStatus.java src/java/org/apache/nutch/api/model/response/NutchStatus.java new file mode 100644 index 0000000..3b494ed --- /dev/null +++ src/java/org/apache/nutch/api/model/response/NutchStatus.java @@ -0,0 +1,61 @@ +/******************************************************************************* + * 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.nutch.api.model.response; + +import java.util.Collection; +import java.util.Date; +import java.util.Set; + +public class NutchStatus { + + private Date startDate; + private Set configuration; + private Collection jobs; + private Collection runningJobs; + + public Date getStartDate() { + return startDate; + } + + public void setStartDate(Date startDate) { + this.startDate = startDate; + } + + public Set getConfiguration() { + return configuration; + } + + public void setConfiguration(Set configuration) { + this.configuration = configuration; + } + + public Collection getJobs() { + return jobs; + } + + public void setJobs(Collection jobs) { + this.jobs = jobs; + } + + public Collection getRunningJobs() { + return runningJobs; + } + + public void setRunningJobs(Collection runningJobs) { + this.runningJobs = runningJobs; + } +} diff --git src/java/org/apache/nutch/api/resources/AbstractResource.java src/java/org/apache/nutch/api/resources/AbstractResource.java new file mode 100644 index 0000000..9e6811e --- /dev/null +++ src/java/org/apache/nutch/api/resources/AbstractResource.java @@ -0,0 +1,48 @@ +/******************************************************************************* + * 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.nutch.api.resources; + +import javax.ws.rs.Produces; +import javax.ws.rs.WebApplicationException; +import javax.ws.rs.core.MediaType; +import javax.ws.rs.core.Response; +import javax.ws.rs.core.Response.Status; + +import org.apache.nutch.api.ConfManager; +import org.apache.nutch.api.JobManager; +import org.apache.nutch.api.NutchServer; +import org.restlet.Context; + +@Produces({ MediaType.APPLICATION_JSON }) +public abstract class AbstractResource { + + protected ConfManager configManager; + protected JobManager jobManager; + protected NutchServer server; + + public AbstractResource() { + server = (NutchServer) Context.getCurrent().getAttributes() + .get(NutchServer.NUTCH_SERVER); + configManager = server.getConfMgr(); + jobManager = server.getJobMgr(); + } + + protected void throwBadRequestException(String message) { + throw new WebApplicationException(Response.status(Status.BAD_REQUEST) + .entity(message).build()); + } +} diff --git src/java/org/apache/nutch/api/resources/AdminResource.java src/java/org/apache/nutch/api/resources/AdminResource.java new file mode 100644 index 0000000..0863b33 --- /dev/null +++ src/java/org/apache/nutch/api/resources/AdminResource.java @@ -0,0 +1,83 @@ +/******************************************************************************* + * 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.nutch.api.resources; + +import java.text.MessageFormat; +import java.util.Date; +import java.util.concurrent.TimeUnit; + +import javax.ws.rs.GET; +import javax.ws.rs.Path; +import javax.ws.rs.QueryParam; + +import org.apache.nutch.api.model.response.NutchStatus; +import org.apache.nutch.api.model.response.JobInfo.State; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +@Path(value = "/admin") +public class AdminResource extends AbstractResource { + private static final int DELAY_SEC = 10; + private static final long DELAY_MILLIS = TimeUnit.SECONDS.toMillis(DELAY_SEC); + + private static final Logger LOG = LoggerFactory + .getLogger(AdminResource.class); + + @GET + @Path("/") + public NutchStatus getNutchStatus() { + NutchStatus status = new NutchStatus(); + + status.setStartDate(new Date(server.getStarted())); + status.setConfiguration(configManager.list()); + status.setJobs(jobManager.list(null, State.ANY)); + status.setRunningJobs(jobManager.list(null, State.RUNNING)); + + return status; + } + + @GET + @Path("/stop") + public String stop(@QueryParam("force") boolean force) { + if (!server.canStop(force)) { + LOG.info("Command 'stop' denied due to unfinished jobs"); + return "Can't stop now. There are jobs running. Try force option."; + } + + scheduleServerStop(); + return MessageFormat.format("Stopping in {0} seconds.", DELAY_SEC); + } + + private void scheduleServerStop() { + LOG.info("Server shutdown scheduled in {} seconds", DELAY_SEC); + Thread thread = new Thread() { + public void run() { + try { + Thread.sleep(DELAY_MILLIS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + server.stop(false); + LOG.info("Service stopped."); + } + }; + thread.setDaemon(true); + thread.start(); + LOG.info("Service shutting down..."); + } + +} diff --git src/java/org/apache/nutch/api/resources/ConfigResource.java src/java/org/apache/nutch/api/resources/ConfigResource.java new file mode 100644 index 0000000..9be0daa --- /dev/null +++ src/java/org/apache/nutch/api/resources/ConfigResource.java @@ -0,0 +1,87 @@ +/******************************************************************************* + * 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.nutch.api.resources; + +import java.util.Map; +import java.util.Set; + +import javax.ws.rs.Consumes; +import javax.ws.rs.DELETE; +import javax.ws.rs.FormParam; +import javax.ws.rs.GET; +import javax.ws.rs.POST; +import javax.ws.rs.PUT; +import javax.ws.rs.Path; +import javax.ws.rs.PathParam; +import javax.ws.rs.WebApplicationException; +import javax.ws.rs.core.MediaType; +import javax.ws.rs.core.Response; +import javax.ws.rs.core.Response.Status; + +import org.apache.nutch.api.model.request.NutchConfig; + +@Path("/config") +public class ConfigResource extends AbstractResource { + public static final String DEFAULT = "default"; + + @GET + @Path("/") + public Set getConfigs() { + return configManager.list(); + } + + @GET + @Path("/{configId}") + public Map getConfig(@PathParam("configId") String configId) { + return configManager.getAsMap(configId); + } + + @GET + @Path("/{configId}/{propertyId}") + public String getProperty(@PathParam("configId") String configId, + @PathParam("propertyId") String propertyId) { + return configManager.getAsMap(configId).get(propertyId); + } + + @DELETE + @Path("/{configId}") + public void deleteConfig(@PathParam("configId") String configId) { + configManager.delete(configId); + } + + @POST + @Path("/{configId}") + @Consumes(MediaType.APPLICATION_JSON) + public String createConfig(NutchConfig newConfig) { + if (newConfig == null) { + throw new WebApplicationException(Response.status(Status.BAD_REQUEST) + .entity("Nutch configuration cannot be empty!").build()); + } + return configManager.create(newConfig); + } + + @PUT + @Path("/{config}/{property}") + public Response update(@PathParam("config") String config, + @PathParam("property") String property, @FormParam("value") String value) { + if (value == null) { + throwBadRequestException("Missing property value!"); + } + configManager.setProperty(config, property, value); + return Response.ok().build(); + } +} diff --git src/java/org/apache/nutch/api/resources/DbResource.java src/java/org/apache/nutch/api/resources/DbResource.java new file mode 100644 index 0000000..7491327 --- /dev/null +++ src/java/org/apache/nutch/api/resources/DbResource.java @@ -0,0 +1,61 @@ +/******************************************************************************* + * 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.nutch.api.resources; + +import java.util.Iterator; +import java.util.Map; +import java.util.WeakHashMap; + +import javax.ws.rs.Consumes; +import javax.ws.rs.POST; +import javax.ws.rs.Path; +import javax.ws.rs.core.MediaType; + +import org.apache.nutch.api.impl.db.DbReader; +import org.apache.nutch.api.model.request.DbFilter; +import org.apache.nutch.api.model.response.DbQueryResult; + +@Path("/db") +public class DbResource extends AbstractResource { + + private Map readers = new WeakHashMap(); + + @POST + @Consumes(MediaType.APPLICATION_JSON) + public DbQueryResult runQuery(DbFilter filter) { + if (filter == null) { + throwBadRequestException("Filter cannot be null!"); + } + + DbQueryResult result = new DbQueryResult(); + Iterator> iterator = getReader().runQuery(filter); + while (iterator.hasNext()) { + result.addValue(iterator.next()); + } + return result; + } + + private DbReader getReader() { + String confId = ConfigResource.DEFAULT; + synchronized (readers) { + if (!readers.containsKey(confId)) { + readers.put(confId, new DbReader(configManager.get(confId), null)); + } + return readers.get(confId); + } + } +} diff --git src/java/org/apache/nutch/api/resources/JobResource.java src/java/org/apache/nutch/api/resources/JobResource.java new file mode 100644 index 0000000..30b14d8 --- /dev/null +++ src/java/org/apache/nutch/api/resources/JobResource.java @@ -0,0 +1,72 @@ +/******************************************************************************* + * 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.nutch.api.resources; + +import java.util.Collection; + +import javax.ws.rs.Consumes; +import javax.ws.rs.GET; +import javax.ws.rs.POST; +import javax.ws.rs.Path; +import javax.ws.rs.PathParam; +import javax.ws.rs.QueryParam; +import javax.ws.rs.core.MediaType; + +import org.apache.nutch.api.model.request.JobConfig; +import org.apache.nutch.api.model.response.JobInfo; +import org.apache.nutch.api.model.response.JobInfo.State; + +@Path(value = "/job") +public class JobResource extends AbstractResource { + + @GET + @Path(value = "/") + public Collection getJobs(@QueryParam("crawlId") String crawlId) { + return jobManager.list(crawlId, State.ANY); + } + + @GET + @Path(value = "/{id}") + public JobInfo getInfo(@PathParam("id") String id, + @QueryParam("crawlId") String crawlId) { + return jobManager.get(crawlId, id); + } + + @GET + @Path(value = "/{id}/stop") + public boolean stop(@PathParam("id") String id, + @QueryParam("crawlId") String crawlId) { + return jobManager.stop(crawlId, id); + } + + @GET + @Path(value = "/{id}/abort") + public boolean abort(@PathParam("id") String id, + @QueryParam("crawlId") String crawlId) { + return jobManager.abort(crawlId, id); + } + + @POST + @Path(value = "/create") + @Consumes(MediaType.APPLICATION_JSON) + public String create(JobConfig config) { + if (config == null) { + throwBadRequestException("Job configuration is required!"); + } + return jobManager.create(config); + } +}