You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@edgent.apache.org by vd...@apache.org on 2016/03/17 21:23:07 UTC
[3/5] incubator-quarks git commit: Job registry and sample monitoring
app
Job registry and sample monitoring app
Project: http://git-wip-us.apache.org/repos/asf/incubator-quarks/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quarks/commit/17b35560
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quarks/tree/17b35560
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quarks/diff/17b35560
Branch: refs/heads/master
Commit: 17b35560612384c718af3540fe50017703224507
Parents: 78f4dad
Author: Victor Dogaru <vd...@apache.org>
Authored: Mon Mar 14 11:11:12 2016 -0700
Committer: Victor Dogaru <vd...@apache.org>
Committed: Thu Mar 17 12:14:22 2016 -0700
----------------------------------------------------------------------
.../quarks/execution/JobRegistryService.java | 95 ++++++++++++
.../java/quarks/runtime/etiao/EtiaoJob.java | 18 +++
.../java/quarks/runtime/etiao/JobRegistry.java | 144 +++++++++++++++++++
.../samples/topology/JobPollingSample.java | 130 +++++++++++++++++
4 files changed, 387 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/17b35560/api/execution/src/main/java/quarks/execution/JobRegistryService.java
----------------------------------------------------------------------
diff --git a/api/execution/src/main/java/quarks/execution/JobRegistryService.java b/api/execution/src/main/java/quarks/execution/JobRegistryService.java
new file mode 100644
index 0000000..53d289b
--- /dev/null
+++ b/api/execution/src/main/java/quarks/execution/JobRegistryService.java
@@ -0,0 +1,95 @@
+/*
+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 quarks.execution;
+
+import java.util.Set;
+
+import quarks.function.BiConsumer;
+
+/**
+ * Contains the methods necessary for the registration and removal
+ * of {@link Job} instances as well as access methods for registered jobs.
+ * Allows clients to register listeners, which are notified on job
+ * registrations, removals, and updates.
+ * <p>
+ * Events are sent to registered listeners when jobs are added, removed,
+ * or their state changes:
+ * <ul>
+ * <li>An {@link EventType#ADD} event type is sent when a job is added.</li>
+ * <li>An {@link EventType#REMOVE} event type is sent when a job is removed.</li>
+ * <li>An {@link EventType#UPDATE} event type is sent when a job is updated.</li>
+ * </ul>
+ */
+public interface JobRegistryService {
+ /**
+ * Job event types.
+ */
+ enum EventType {
+ /** A Job has been added to the registry. */
+ ADD,
+ /** A Job has been removed from the registry. */
+ REMOVE,
+ /** A registered Job has been updated. */
+ UPDATE
+ }
+
+ /**
+ * Adds a handler to a collection of listeners that will be notified
+ * on job registration and state changes. Listeners will be notified
+ * in the order in which they are added.
+ * <p>
+ * A listener is notified of all existing jobs when it is first added.
+ *
+ * @param listener the listener that will be added
+ */
+ void addListener(BiConsumer<JobRegistryService.EventType, Job> listener);
+
+ /**
+ * Removes a handler from this registry's collection of listeners.
+ *
+ * @param listener the listener that will be removed
+ */
+ void removeListener(
+ BiConsumer<JobRegistryService.EventType, Job> listener);
+
+ /**
+ * Returns a set of all the registered job identifiers.
+ *
+ * @return the identifiers of all the jobs
+ */
+ Set<String> getJobIds();
+
+ /**
+ * Returns a job given its identifier.
+ *
+ * @return the job or {@code null} if no job is registered with that
+ * identifier.
+ */
+ Job getJob(String id);
+
+ /**
+ * Removes the job specified by the given identifier.
+ *
+ * @param jobId the identifier of the job to remove
+ * @return whether or not the job was removed
+ */
+ boolean removeJob(String jobId);
+
+ // TODO add job retrieval given its name
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/17b35560/runtime/etiao/src/main/java/quarks/runtime/etiao/EtiaoJob.java
----------------------------------------------------------------------
diff --git a/runtime/etiao/src/main/java/quarks/runtime/etiao/EtiaoJob.java b/runtime/etiao/src/main/java/quarks/runtime/etiao/EtiaoJob.java
index 3f6e36a..bf83eb9 100644
--- a/runtime/etiao/src/main/java/quarks/runtime/etiao/EtiaoJob.java
+++ b/runtime/etiao/src/main/java/quarks/runtime/etiao/EtiaoJob.java
@@ -11,6 +11,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
+import quarks.execution.JobRegistryService;
import quarks.execution.mbeans.JobMXBean;
import quarks.execution.services.ControlService;
import quarks.execution.services.ServiceContainer;
@@ -31,6 +32,7 @@ public class EtiaoJob extends AbstractGraphJob implements JobContext {
private final String topologyName;
private String name;
private final ServiceContainer containerServices;
+ private final JobRegistry jobs;
private static final AtomicInteger jobID = new AtomicInteger(0);
@@ -51,6 +53,10 @@ public class EtiaoJob extends AbstractGraphJob implements JobContext {
ControlService cs = container.getService(ControlService.class);
if (cs != null)
cs.registerControl(JobMXBean.TYPE, getId(), getName(), JobMXBean.class, new EtiaoJobBean(this));
+
+ this.jobs = (JobRegistry) container.getService(JobRegistryService.class);
+ if (jobs != null)
+ jobs.add(this);
}
/**
@@ -125,6 +131,7 @@ public class EtiaoJob extends AbstractGraphJob implements JobContext {
throw new IllegalArgumentException(cause.name());
else {
setNextState(desiredState);
+ updateRegistry();
return getCurrentState();
}
}
@@ -133,6 +140,11 @@ public class EtiaoJob extends AbstractGraphJob implements JobContext {
return !inTransition() && stateMap.get(getCurrentState()).contains(desiredState);
}
+ protected synchronized void completeTransition() {
+ super.completeTransition();
+ updateRegistry();
+ }
+
void onActionComplete() {
completeTransition();
}
@@ -171,5 +183,11 @@ public class EtiaoJob extends AbstractGraphJob implements JobContext {
public void setName(String name) {
this.name = name;
+ updateRegistry();
+ }
+
+ private void updateRegistry() {
+ if (jobs != null)
+ jobs.update(this);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/17b35560/runtime/etiao/src/main/java/quarks/runtime/etiao/JobRegistry.java
----------------------------------------------------------------------
diff --git a/runtime/etiao/src/main/java/quarks/runtime/etiao/JobRegistry.java b/runtime/etiao/src/main/java/quarks/runtime/etiao/JobRegistry.java
new file mode 100644
index 0000000..f163bcc
--- /dev/null
+++ b/runtime/etiao/src/main/java/quarks/runtime/etiao/JobRegistry.java
@@ -0,0 +1,144 @@
+/*
+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 quarks.runtime.etiao;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import quarks.execution.Job;
+import quarks.execution.JobRegistryService;
+import quarks.function.BiConsumer;
+
+/**
+ * Maintains a set of registered jobs and a set of listeners.
+ * Notifies listeners on job additions, deletions and updates.
+ */
+public class JobRegistry implements JobRegistryService {
+ private final ConcurrentMap<String /*JobId*/, Job> jobs;
+ private final Broadcaster<JobRegistryService.EventType, Job> listeners;
+
+ /**
+ * Job event types.
+ */
+ public enum EventType {
+ /** A Job has been added to the registry. */
+ ADD,
+ /** A Job has been removed from the registry. */
+ REMOVE,
+ /** A registered Job has been updated. */
+ UPDATE
+ }
+
+ /**
+ * Creates a new {@link JobRegistry}.
+ */
+ public JobRegistry() {
+ this.jobs = new ConcurrentHashMap<String, Job>();
+ this.listeners = new Broadcaster<JobRegistryService.EventType, Job>();
+ }
+
+ @Override
+ public void addListener(BiConsumer<JobRegistryService.EventType, Job> listener) {
+ listeners.add(listener);
+
+ jobs.forEach((k, v) -> {
+ listener.accept(JobRegistryService.EventType.ADD, v);
+ });
+ }
+
+ @Override
+ public void removeListener(BiConsumer<JobRegistryService.EventType, Job> listener) {
+ listeners.remove(listener);
+ }
+
+ @Override
+ public Set<String> getJobIds() {
+ return Collections.unmodifiableSet(jobs.keySet());
+ }
+
+ @Override
+ public Job getJob(String id) {
+ return jobs.get(id);
+ }
+
+ /**
+ * Adds the specified {@link Job} under the given name.
+ *
+ * @param job the job to register
+ *
+ * @throws IllegalArgumentException if a job with the same job name
+ * is already registered
+ */
+ void add(Job job) throws IllegalArgumentException {
+ final Job existing = jobs.putIfAbsent(job.getId(), job);
+ if (existing == null) {
+ listeners.onEvent(JobRegistryService.EventType.ADD, job);
+ } else {
+ throw new IllegalArgumentException("A job with Id " + job.getId()
+ + " already exists");
+ }
+ }
+
+ @Override
+ public boolean removeJob(String jobId) {
+ final Job removed = jobs.remove(jobId);
+ if (removed != null) {
+ listeners.onEvent(JobRegistryService.EventType.REMOVE, removed);
+ return true;
+ }
+ return false;
+ }
+
+ /**
+ * Notifies listeners that the specified job has been updated.
+ *
+ * @param job the job
+ */
+ void update(Job job) {
+ if (jobs.containsValue(job)) {
+ listeners.onEvent(JobRegistryService.EventType.UPDATE, job);
+ }
+ }
+
+ private static class Broadcaster<T, O> {
+ private final List<BiConsumer<T, O>> listeners;
+
+ Broadcaster() {
+ this.listeners = new CopyOnWriteArrayList<BiConsumer<T, O>>();
+ }
+
+ void add(BiConsumer<T, O> listener) {
+ listeners.add(listener);
+ }
+
+ void remove(BiConsumer<T, O> listener) {
+ listeners.remove(listener);
+ }
+
+ private void onEvent(T event, O job) {
+ for (BiConsumer<T, O> listener : listeners) {
+ listener.accept(event, job);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/17b35560/samples/topology/src/main/java/quarks/samples/topology/JobPollingSample.java
----------------------------------------------------------------------
diff --git a/samples/topology/src/main/java/quarks/samples/topology/JobPollingSample.java b/samples/topology/src/main/java/quarks/samples/topology/JobPollingSample.java
new file mode 100644
index 0000000..4988210
--- /dev/null
+++ b/samples/topology/src/main/java/quarks/samples/topology/JobPollingSample.java
@@ -0,0 +1,130 @@
+/*
+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 quarks.samples.topology;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import quarks.execution.Job;
+import quarks.execution.JobRegistryService;
+import quarks.providers.development.DevelopmentProvider;
+import quarks.runtime.etiao.JobRegistry;
+import quarks.topology.TStream;
+import quarks.topology.Topology;
+
+/**
+ * Job monitoring by polling job state.
+ * <p>
+ * Demonstrates job monitoring using the {@link JobRegistryService} service.
+ * The example starts a system monitoring application then concurrently
+ * submits two jobs. The monitoring application is using a polling source
+ * to periodically scan the job registry and generates tuples containing the
+ * current state of registered jobs. Tuples are pushed to a sink which prints
+ * them onto the system output.
+ */
+public class JobPollingSample {
+ private final DevelopmentProvider dtp;
+
+ public static void main(String[] args) throws Exception {
+
+ JobPollingSample app = new JobPollingSample();
+
+ // Start monitoring app
+ app.monitor();
+
+ Thread.sleep(2000);
+
+ // Asynchronously start two jobs
+ ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);
+ executor.schedule(app.getCallable("Monitored1"), 100, TimeUnit.MILLISECONDS);
+ executor.schedule(app.getCallable("Monitored2"), 3300, TimeUnit.MILLISECONDS);
+ }
+
+ JobPollingSample() throws Exception {
+ this.dtp = new DevelopmentProvider();
+ dtp.getServices().addService(JobRegistryService.class, new JobRegistry());
+ }
+
+ void monitored(String name) throws InterruptedException, ExecutionException {
+ Topology t = dtp.newTopology(name);
+
+ Random r = new Random();
+ TStream<Double> d = t.poll(() -> r.nextGaussian(), 100, TimeUnit.MILLISECONDS);
+ d.sink(tuple -> System.out.print("."));
+
+ Thread.sleep(2000);
+ Future<Job> f = dtp.submit(t);
+ Job job = f.get();
+ Thread.sleep(5000);
+ job.stateChange(Job.Action.CLOSE);
+ Thread.sleep(2000);
+
+ provider().getServices().getService(JobRegistryService.class).removeJob(job.getId());
+ }
+
+ /**
+ * Monitoring application polls the job registry every 1 sec.
+ */
+ void monitor() {
+ Topology t = dtp.newTopology("Monitor");
+
+ TStream<Job.State[]> state = t.poll(() -> {
+ JobRegistryService jobs = provider().getServices().getService(JobRegistryService.class);
+ List<Job.State> states = new ArrayList<>();
+ if (jobs != null) {
+ for (String id: jobs.getJobIds()) {
+ states.add(jobs.getJob(id).getCurrentState());
+ }
+ }
+ return states.toArray(new Job.State[0]);
+ }, 1, TimeUnit.SECONDS);
+
+ state.sink(states -> {
+ StringBuffer sb = new StringBuffer();
+ for (Job.State s : states) {
+ sb.append(s).append(',');
+ }
+ System.out.println(sb.toString());
+ });
+
+ dtp.submit(t);
+ }
+
+ private DevelopmentProvider provider() {
+ return dtp;
+ }
+
+ private Callable<Integer> getCallable(String name) {
+ return new Callable<Integer>() {
+
+ @Override
+ public Integer call() throws Exception {
+ monitored(name);
+ return new Integer(0);
+ }
+ };
+ }
+}