You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@brooklyn.apache.org by he...@apache.org on 2014/09/26 00:49:31 UTC
[07/12] Convert Feed to extend EntityAdjunct
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/159745bf/core/src/main/java/brooklyn/event/feed/shell/ShellFeed.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/event/feed/shell/ShellFeed.java b/core/src/main/java/brooklyn/event/feed/shell/ShellFeed.java
index b90b2ea..df088b8 100644
--- a/core/src/main/java/brooklyn/event/feed/shell/ShellFeed.java
+++ b/core/src/main/java/brooklyn/event/feed/shell/ShellFeed.java
@@ -30,6 +30,8 @@ import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import brooklyn.config.ConfigKey;
+import brooklyn.entity.basic.ConfigKeys;
import brooklyn.entity.basic.EntityInternal;
import brooklyn.entity.basic.EntityLocal;
import brooklyn.event.feed.AbstractFeed;
@@ -50,6 +52,7 @@ import com.google.common.collect.HashMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.SetMultimap;
import com.google.common.collect.Sets;
+import com.google.common.reflect.TypeToken;
/**
* Provides a feed of attribute values, by executing shell commands (on the local machine where
@@ -96,6 +99,11 @@ public class ShellFeed extends AbstractFeed {
public static final Logger log = LoggerFactory.getLogger(ShellFeed.class);
+ @SuppressWarnings("serial")
+ private static final ConfigKey<SetMultimap<ShellPollIdentifier, ShellPollConfig<?>>> POLLS = ConfigKeys.newConfigKey(
+ new TypeToken<SetMultimap<ShellPollIdentifier, ShellPollConfig<?>>>() {},
+ "polls");
+
public static Builder builder() {
return new Builder();
}
@@ -126,6 +134,7 @@ public class ShellFeed extends AbstractFeed {
public ShellFeed build() {
built = true;
ShellFeed result = new ShellFeed(this);
+ result.setEntity(checkNotNull(entity, "entity"));
result.start();
return result;
}
@@ -171,12 +180,16 @@ public class ShellFeed extends AbstractFeed {
}
}
- // Treat as immutable once built
- private final SetMultimap<ShellPollIdentifier, ShellPollConfig<?>> polls = HashMultimap.<ShellPollIdentifier,ShellPollConfig<?>>create();
-
+ /**
+ * For rebind; do not call directly; use builder
+ */
+ public ShellFeed() {
+ }
+
protected ShellFeed(Builder builder) {
- super(builder.entity);
-
+ super();
+
+ SetMultimap<ShellPollIdentifier, ShellPollConfig<?>> polls = HashMultimap.<ShellPollIdentifier,ShellPollConfig<?>>create();
for (ShellPollConfig<?> config : builder.polls) {
@SuppressWarnings({ "unchecked", "rawtypes" })
ShellPollConfig<?> configCopy = new ShellPollConfig(config);
@@ -190,10 +203,13 @@ public class ShellFeed extends AbstractFeed {
polls.put(new ShellPollIdentifier(command, env, dir, input, context, timeout), configCopy);
}
+ setConfig(POLLS, polls);
}
@Override
protected void preStart() {
+ SetMultimap<ShellPollIdentifier, ShellPollConfig<?>> polls = getConfig(POLLS);
+
for (final ShellPollIdentifier pollInfo : polls.keySet()) {
Set<ShellPollConfig<?>> configs = polls.get(pollInfo);
long minPeriod = Integer.MAX_VALUE;
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/159745bf/core/src/main/java/brooklyn/event/feed/ssh/SshFeed.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/event/feed/ssh/SshFeed.java b/core/src/main/java/brooklyn/event/feed/ssh/SshFeed.java
index 14988ab..5b41d19 100644
--- a/core/src/main/java/brooklyn/event/feed/ssh/SshFeed.java
+++ b/core/src/main/java/brooklyn/event/feed/ssh/SshFeed.java
@@ -31,7 +31,9 @@ import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import brooklyn.config.ConfigKey;
import brooklyn.entity.Entity;
+import brooklyn.entity.basic.ConfigKeys;
import brooklyn.entity.basic.EntityLocal;
import brooklyn.event.feed.AbstractFeed;
import brooklyn.event.feed.AttributePollHandler;
@@ -52,6 +54,7 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import com.google.common.collect.SetMultimap;
import com.google.common.collect.Sets;
+import com.google.common.reflect.TypeToken;
/**
* Provides a feed of attribute values, by polling over ssh.
@@ -90,7 +93,19 @@ import com.google.common.collect.Sets;
public class SshFeed extends AbstractFeed {
public static final Logger log = LoggerFactory.getLogger(SshFeed.class);
-
+
+ @SuppressWarnings("serial")
+ public static final ConfigKey<Supplier<SshMachineLocation>> MACHINE = ConfigKeys.newConfigKey(
+ new TypeToken<Supplier<SshMachineLocation>>() {},
+ "machine");
+
+ public static final ConfigKey<Boolean> EXEC_AS_COMMAND = ConfigKeys.newBooleanConfigKey("execAsCommand");
+
+ @SuppressWarnings("serial")
+ public static final ConfigKey<SetMultimap<SshPollIdentifier, SshPollConfig<?>>> POLLS = ConfigKeys.newConfigKey(
+ new TypeToken<SetMultimap<SshPollIdentifier, SshPollConfig<?>>>() {},
+ "polls");
+
public static Builder builder() {
return new Builder();
}
@@ -99,8 +114,7 @@ public class SshFeed extends AbstractFeed {
private EntityLocal entity;
private boolean onlyIfServiceUp = false;
private Supplier<SshMachineLocation> machine;
- private long period = 500;
- private TimeUnit periodUnits = TimeUnit.MILLISECONDS;
+ private Duration period = Duration.of(500, TimeUnit.MILLISECONDS);
private List<SshPollConfig<?>> polls = Lists.newArrayList();
private boolean execAsCommand = false;
private volatile boolean built;
@@ -122,15 +136,14 @@ public class SshFeed extends AbstractFeed {
return this;
}
public Builder period(Duration period) {
- return period(period.toMilliseconds(), TimeUnit.MILLISECONDS);
+ this.period = period;
+ return this;
}
public Builder period(long millis) {
- return period(millis, TimeUnit.MILLISECONDS);
+ return period(Duration.of(millis, TimeUnit.MILLISECONDS));
}
public Builder period(long val, TimeUnit units) {
- this.period = val;
- this.periodUnits = units;
- return this;
+ return period(Duration.of(val, units));
}
public Builder poll(SshPollConfig<?> config) {
polls.add(config);
@@ -147,6 +160,7 @@ public class SshFeed extends AbstractFeed {
public SshFeed build() {
built = true;
SshFeed result = new SshFeed(this);
+ result.setEntity(checkNotNull(entity, "entity"));
result.start();
return result;
}
@@ -181,39 +195,46 @@ public class SshFeed extends AbstractFeed {
}
}
- private final Supplier<SshMachineLocation> machine;
- private final boolean execAsCommand;
-
- // Treat as immutable once built
- private final SetMultimap<SshPollIdentifier, SshPollConfig<?>> polls = HashMultimap.<SshPollIdentifier,SshPollConfig<?>>create();
-
/** @deprecated since 0.7.0, use static convenience on {@link Locations} */
@Deprecated
public static SshMachineLocation getMachineOfEntity(Entity entity) {
return Machines.findUniqueSshMachineLocation(entity.getLocations()).orNull();
}
+ /**
+ * For rebind; do not call directly; use builder
+ */
+ public SshFeed() {
+ }
+
protected SshFeed(final Builder builder) {
- super(builder.entity, builder.onlyIfServiceUp);
- machine = builder.machine != null ? builder.machine :
- new Supplier<SshMachineLocation>() {
- @Override
- public SshMachineLocation get() {
- return Locations.findUniqueSshMachineLocation(entity.getLocations()).get();
- }
- };
- execAsCommand = builder.execAsCommand;
+ setConfig(ONLY_IF_SERVICE_UP, builder.onlyIfServiceUp);
+ setConfig(MACHINE, builder.machine != null ? builder.machine : null);
+ setConfig(EXEC_AS_COMMAND, builder.execAsCommand);
+ SetMultimap<SshPollIdentifier, SshPollConfig<?>> polls = HashMultimap.<SshPollIdentifier,SshPollConfig<?>>create();
for (SshPollConfig<?> config : builder.polls) {
@SuppressWarnings({ "unchecked", "rawtypes" })
SshPollConfig<?> configCopy = new SshPollConfig(config);
- if (configCopy.getPeriod() < 0) configCopy.period(builder.period, builder.periodUnits);
+ if (configCopy.getPeriod() < 0) configCopy.period(builder.period);
polls.put(new SshPollIdentifier(config.getCommandSupplier(), config.getEnvSupplier()), configCopy);
}
+ setConfig(POLLS, polls);
}
+ protected SshMachineLocation getMachine() {
+ Supplier<SshMachineLocation> supplier = getConfig(MACHINE);
+ if (supplier != null) {
+ return supplier.get();
+ } else {
+ return Locations.findUniqueSshMachineLocation(entity.getLocations()).get();
+ }
+ }
+
@Override
protected void preStart() {
+ SetMultimap<SshPollIdentifier, SshPollConfig<?>> polls = getConfig(POLLS);
+
for (final SshPollIdentifier pollInfo : polls.keySet()) {
Set<SshPollConfig<?>> configs = polls.get(pollInfo);
long minPeriod = Integer.MAX_VALUE;
@@ -240,6 +261,8 @@ public class SshFeed extends AbstractFeed {
}
private SshPollValue exec(String command, Map<String,String> env) throws IOException {
+ SshMachineLocation machine = getMachine();
+ Boolean execAsCommand = getConfig(EXEC_AS_COMMAND);
if (log.isTraceEnabled()) log.trace("Ssh polling for {}, executing {} with env {}", new Object[] {machine, command, env});
ByteArrayOutputStream stdout = new ByteArrayOutputStream();
ByteArrayOutputStream stderr = new ByteArrayOutputStream();
@@ -249,14 +272,14 @@ public class SshFeed extends AbstractFeed {
.configure(SshTool.PROP_NO_EXTRA_OUTPUT, true)
.configure(SshTool.PROP_OUT_STREAM, stdout)
.configure(SshTool.PROP_ERR_STREAM, stderr);
- if (execAsCommand) {
- exitStatus = machine.get().execCommands(flags.getAllConfig(),
+ if (Boolean.TRUE.equals(execAsCommand)) {
+ exitStatus = machine.execCommands(flags.getAllConfig(),
"ssh-feed", ImmutableList.of(command), env);
} else {
- exitStatus = machine.get().execScript(flags.getAllConfig(),
+ exitStatus = machine.execScript(flags.getAllConfig(),
"ssh-feed", ImmutableList.of(command), env);
}
- return new SshPollValue(machine.get(), exitStatus, new String(stdout.toByteArray()), new String(stderr.toByteArray()));
+ return new SshPollValue(machine, exitStatus, new String(stdout.toByteArray()), new String(stderr.toByteArray()));
}
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/159745bf/core/src/main/java/brooklyn/event/feed/windows/WindowsPerformanceCounterFeed.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/event/feed/windows/WindowsPerformanceCounterFeed.java b/core/src/main/java/brooklyn/event/feed/windows/WindowsPerformanceCounterFeed.java
index 7ccc9fc..c3bfeb3 100644
--- a/core/src/main/java/brooklyn/event/feed/windows/WindowsPerformanceCounterFeed.java
+++ b/core/src/main/java/brooklyn/event/feed/windows/WindowsPerformanceCounterFeed.java
@@ -26,6 +26,7 @@ import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.SortedMap;
+import java.util.SortedSet;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
@@ -36,6 +37,8 @@ import javax.annotation.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import brooklyn.config.ConfigKey;
+import brooklyn.entity.basic.ConfigKeys;
import brooklyn.entity.basic.EntityInternal;
import brooklyn.entity.basic.EntityLocal;
import brooklyn.entity.effector.EffectorTasks;
@@ -60,7 +63,8 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
-import com.google.common.collect.Ordering;
+import com.google.common.collect.Sets;
+import com.google.common.reflect.TypeToken;
/**
* A sensor feed that retrieves performance counters from a Windows host and posts the values to sensors.
@@ -95,44 +99,52 @@ public class WindowsPerformanceCounterFeed extends AbstractFeed {
protected static final Pattern lineWithPerfData = Pattern.compile("^\"[\\d:/\\-. ]+\",\".*\"$", Pattern.MULTILINE);
private static final Joiner JOINER_ON_SPACE = Joiner.on(' ');
+ @SuppressWarnings("serial")
+ public static final ConfigKey<Set<WindowsPerformanceCounterPollConfig<?>>> POLLS = ConfigKeys.newConfigKey(
+ new TypeToken<Set<WindowsPerformanceCounterPollConfig<?>>>() {},
+ "polls");
+
public static Builder builder() {
return new Builder();
}
public static class Builder {
private EntityLocal entity;
- private SortedMap<String, AttributeSensor> sensors = Maps.newTreeMap(Ordering.natural());
- private long period = 30;
- private TimeUnit periodUnits = TimeUnit.SECONDS;
+ private Set<WindowsPerformanceCounterPollConfig<?>> polls = Sets.newLinkedHashSet();
+ private Duration period = Duration.of(30, TimeUnit.SECONDS);
private volatile boolean built;
public Builder entity(EntityLocal val) {
this.entity = checkNotNull(val, "entity");
return this;
}
- public Builder addSensor(String performanceCounterName, AttributeSensor sensor) {
- sensors.put(checkNotNull(performanceCounterName, "performanceCounterName"),
- checkNotNull(sensor, "sensor"));
+ public Builder addSensor(WindowsPerformanceCounterPollConfig<?> config) {
+ polls.add(config);
return this;
}
+ public Builder addSensor(String performanceCounterName, AttributeSensor<?> sensor) {
+ return addSensor(new WindowsPerformanceCounterPollConfig(sensor).performanceCounterName(checkNotNull(performanceCounterName, "performanceCounterName")));
+ }
public Builder addSensors(Map<String, AttributeSensor> sensors) {
- sensors.putAll(checkNotNull(sensors, "sensors"));
+ for (Map.Entry<String, AttributeSensor> entry : sensors.entrySet()) {
+ addSensor(entry.getKey(), entry.getValue());
+ }
return this;
}
public Builder period(Duration period) {
- return period(checkNotNull(period, "period").toMilliseconds(), TimeUnit.MILLISECONDS);
+ period = checkNotNull(period, "period");
+ return this;
}
public Builder period(long millis) {
return period(millis, TimeUnit.MILLISECONDS);
}
public Builder period(long val, TimeUnit units) {
- this.period = val;
- this.periodUnits = checkNotNull(units, "units");
- return this;
+ return period(Duration.of(val, units));
}
public WindowsPerformanceCounterFeed build() {
built = true;
WindowsPerformanceCounterFeed result = new WindowsPerformanceCounterFeed(this);
+ result.setEntity(checkNotNull(entity, "entity"));
result.start();
return result;
}
@@ -142,34 +154,48 @@ public class WindowsPerformanceCounterFeed extends AbstractFeed {
}
}
- private final EntityLocal entity;
- private final long period;
- private final TimeUnit periodUnits;
- private final SortedMap<String, AttributeSensor> attributeSensors;
- private final ProcessTaskFactory<Integer> taskFactory;
+ /**
+ * For rebind; do not call directly; use builder
+ */
+ public WindowsPerformanceCounterFeed() {
+ }
protected WindowsPerformanceCounterFeed(Builder builder) {
- super(checkNotNull(builder.entity, "builder.entity"));
- entity = builder.entity;
- period = builder.period;
- periodUnits = builder.periodUnits;
- attributeSensors = builder.sensors;
+ Set<WindowsPerformanceCounterPollConfig<?>> polls = Sets.newLinkedHashSet();
+ for (WindowsPerformanceCounterPollConfig<?> config : builder.polls) {
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ WindowsPerformanceCounterPollConfig<?> configCopy = new WindowsPerformanceCounterPollConfig(config);
+ if (configCopy.getPeriod() < 0) configCopy.period(builder.period);
+ polls.add(configCopy);
+ }
+ setConfig(POLLS, polls);
+ }
+
+ @Override
+ protected void preStart() {
+ Set<WindowsPerformanceCounterPollConfig<?>> polls = getConfig(POLLS);
+
+ long minPeriod = Integer.MAX_VALUE;
+ SortedSet<String> performanceCounterNames = Sets.newTreeSet();
+ for (WindowsPerformanceCounterPollConfig<?> config : polls) {
+ minPeriod = Math.min(minPeriod, config.getPeriod());
+ performanceCounterNames.add(config.getPerformanceCounterName());
+ }
+
SshMachineLocation machine = EffectorTasks.getSshMachine(getEntity());
Iterable<String> allParams = ImmutableList.<String>builder()
.add("typeperf")
- .addAll(Iterables.transform(attributeSensors.keySet(), QuoteStringFunction.INSTANCE))
+ .addAll(Iterables.transform(performanceCounterNames, QuoteStringFunction.INSTANCE))
.add("-sc")
.add("1")
.build();
String command = JOINER_ON_SPACE.join(allParams);
log.debug("Windows performance counter poll command will be: {}", command);
- taskFactory = SshTasks.newSshExecTaskFactory(machine, command)
+
+ final ProcessTaskFactory<Integer> taskFactory = SshTasks.newSshExecTaskFactory(machine, command)
.allowingNonZeroExitCode()
.runAsCommand();
- }
- @Override
- protected void preStart() {
final Callable<SshPollValue> queryForCounterValues = new Callable<SshPollValue>() {
public SshPollValue call() throws Exception {
ProcessTaskWrapper<Integer> taskWrapper = taskFactory.newTask();
@@ -184,8 +210,8 @@ public class WindowsPerformanceCounterFeed extends AbstractFeed {
getPoller().scheduleAtFixedRate(
new CallInEntityExecutionContext<SshPollValue>(entity, queryForCounterValues),
- new SendPerfCountersToSensors(entity, attributeSensors),
- periodUnits.toMillis(period));
+ new SendPerfCountersToSensors(entity, polls),
+ minPeriod);
}
@SuppressWarnings("unchecked")
@@ -227,12 +253,16 @@ public class WindowsPerformanceCounterFeed extends AbstractFeed {
private static final Set<? extends Class<? extends Number>> INTEGER_TYPES
= ImmutableSet.of(Integer.class, Long.class, Byte.class, Short.class, BigInteger.class);
- private SortedMap<String, AttributeSensor> sensorMap;
- private EntityLocal entity;
+ private final EntityLocal entity;
+ private final SortedMap<String, AttributeSensor> sensorMap;
- public SendPerfCountersToSensors(EntityLocal entity, SortedMap<String, AttributeSensor> sensorMap) {
- this.sensorMap = sensorMap;
+ public SendPerfCountersToSensors(EntityLocal entity, Set<WindowsPerformanceCounterPollConfig<?>> polls) {
this.entity = entity;
+
+ sensorMap = Maps.newTreeMap();
+ for (WindowsPerformanceCounterPollConfig<?> config : polls) {
+ sensorMap.put(config.getPerformanceCounterName(), config.getSensor());
+ }
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/159745bf/core/src/main/java/brooklyn/event/feed/windows/WindowsPerformanceCounterPollConfig.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/event/feed/windows/WindowsPerformanceCounterPollConfig.java b/core/src/main/java/brooklyn/event/feed/windows/WindowsPerformanceCounterPollConfig.java
new file mode 100644
index 0000000..fe4f897
--- /dev/null
+++ b/core/src/main/java/brooklyn/event/feed/windows/WindowsPerformanceCounterPollConfig.java
@@ -0,0 +1,54 @@
+/*
+ * 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 brooklyn.event.feed.windows;
+
+import brooklyn.event.AttributeSensor;
+import brooklyn.event.feed.PollConfig;
+
+import com.google.common.base.Function;
+import com.google.common.base.Functions;
+
+public class WindowsPerformanceCounterPollConfig<T> extends PollConfig<Object, T, WindowsPerformanceCounterPollConfig<T>>{
+
+ private String performanceCounterName;
+
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ public WindowsPerformanceCounterPollConfig(AttributeSensor<T> sensor) {
+ super(sensor);
+ onSuccess((Function)Functions.identity());
+ }
+
+ public WindowsPerformanceCounterPollConfig(WindowsPerformanceCounterPollConfig<T> other) {
+ super(other);
+ this.performanceCounterName = other.performanceCounterName;
+ }
+
+ public String getPerformanceCounterName() {
+ return performanceCounterName;
+ }
+
+ public WindowsPerformanceCounterPollConfig<T> performanceCounterName(String val) {
+ this.performanceCounterName = val; return this;
+ }
+
+ @Override
+ public String toString() {
+ return "windowsPerformanceCounter["+performanceCounterName+"]";
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/159745bf/core/src/main/java/brooklyn/internal/BrooklynFeatureEnablement.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/internal/BrooklynFeatureEnablement.java b/core/src/main/java/brooklyn/internal/BrooklynFeatureEnablement.java
index c16c948..d04b734 100644
--- a/core/src/main/java/brooklyn/internal/BrooklynFeatureEnablement.java
+++ b/core/src/main/java/brooklyn/internal/BrooklynFeatureEnablement.java
@@ -41,6 +41,8 @@ public class BrooklynFeatureEnablement {
public static final String FEATURE_ENRICHER_PERSISTENCE_PROPERTY = "brooklyn.experimental.feature.enricherPersistence";
+ public static final String FEATURE_FEED_PERSISTENCE_PROPERTY = "brooklyn.experimental.feature.feedPersistence";
+
private static final Map<String, Boolean> FEATURE_ENABLEMENTS = Maps.newLinkedHashMap();
private static final Object MUTEX = new Object();
@@ -52,6 +54,7 @@ public class BrooklynFeatureEnablement {
setDefault(FEATURE_POLICY_PERSISTENCE_PROPERTY, true);
setDefault(FEATURE_ENRICHER_PERSISTENCE_PROPERTY, true);
+ setDefault(FEATURE_FEED_PERSISTENCE_PROPERTY, true);
}
static {
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/159745bf/core/src/main/java/brooklyn/management/internal/EntityChangeListener.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/management/internal/EntityChangeListener.java b/core/src/main/java/brooklyn/management/internal/EntityChangeListener.java
index 8e1dbe6..811092a 100644
--- a/core/src/main/java/brooklyn/management/internal/EntityChangeListener.java
+++ b/core/src/main/java/brooklyn/management/internal/EntityChangeListener.java
@@ -20,6 +20,7 @@ package brooklyn.management.internal;
import brooklyn.config.ConfigKey;
import brooklyn.entity.Effector;
+import brooklyn.entity.Feed;
import brooklyn.event.AttributeSensor;
import brooklyn.policy.Enricher;
import brooklyn.policy.Policy;
@@ -39,6 +40,8 @@ public interface EntityChangeListener {
@Override public void onPolicyRemoved(Policy policy) {}
@Override public void onEnricherAdded(Enricher enricher) {}
@Override public void onEnricherRemoved(Enricher enricher) {}
+ @Override public void onFeedAdded(Feed feed) {}
+ @Override public void onFeedRemoved(Feed feed) {}
@Override public void onEffectorStarting(Effector<?> effector) {}
@Override public void onEffectorCompleted(Effector<?> effector) {}
};
@@ -65,6 +68,10 @@ public interface EntityChangeListener {
void onEnricherRemoved(Enricher enricher);
+ void onFeedAdded(Feed feed);
+
+ void onFeedRemoved(Feed feed);
+
void onEffectorStarting(Effector<?> effector);
void onEffectorCompleted(Effector<?> effector);
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/159745bf/core/src/main/java/brooklyn/management/internal/EntityManagementSupport.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/brooklyn/management/internal/EntityManagementSupport.java b/core/src/main/java/brooklyn/management/internal/EntityManagementSupport.java
index e7a4bbe..b820cdc 100644
--- a/core/src/main/java/brooklyn/management/internal/EntityManagementSupport.java
+++ b/core/src/main/java/brooklyn/management/internal/EntityManagementSupport.java
@@ -30,6 +30,7 @@ import brooklyn.config.ConfigKey;
import brooklyn.entity.Application;
import brooklyn.entity.Effector;
import brooklyn.entity.Entity;
+import brooklyn.entity.Feed;
import brooklyn.entity.basic.AbstractEntity;
import brooklyn.entity.basic.Entities;
import brooklyn.entity.basic.EntityInternal;
@@ -379,6 +380,11 @@ public class EntityManagementSupport {
getManagementContext().getRebindManager().getChangeListener().onManaged(enricher);
}
@Override
+ public void onFeedAdded(Feed feed) {
+ getManagementContext().getRebindManager().getChangeListener().onChanged(entity);
+ getManagementContext().getRebindManager().getChangeListener().onManaged(feed);
+ }
+ @Override
public void onPolicyRemoved(Policy policy) {
getManagementContext().getRebindManager().getChangeListener().onChanged(entity);
getManagementContext().getRebindManager().getChangeListener().onUnmanaged(policy);
@@ -389,6 +395,11 @@ public class EntityManagementSupport {
getManagementContext().getRebindManager().getChangeListener().onUnmanaged(enricher);
}
@Override
+ public void onFeedRemoved(Feed feed) {
+ getManagementContext().getRebindManager().getChangeListener().onChanged(entity);
+ getManagementContext().getRebindManager().getChangeListener().onUnmanaged(feed);
+ }
+ @Override
public void onAttributeChanged(AttributeSensor<?> attribute) {
// TODO Could make this more efficient by inspecting the attribute to decide if needs persisted
// immediately, or not important, or transient (e.g. do we really need to persist
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/159745bf/core/src/test/java/brooklyn/entity/basic/EntityFunctionsTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/entity/basic/EntityFunctionsTest.java b/core/src/test/java/brooklyn/entity/basic/EntityFunctionsTest.java
index 106d3b3..02fcd9b 100644
--- a/core/src/test/java/brooklyn/entity/basic/EntityFunctionsTest.java
+++ b/core/src/test/java/brooklyn/entity/basic/EntityFunctionsTest.java
@@ -26,17 +26,23 @@ import org.testng.annotations.Test;
import brooklyn.entity.BrooklynAppUnitTestSupport;
import brooklyn.entity.proxying.EntitySpec;
+import brooklyn.location.Location;
import brooklyn.test.entity.TestEntity;
+import com.google.common.base.Predicates;
+import com.google.common.collect.ImmutableList;
+
public class EntityFunctionsTest extends BrooklynAppUnitTestSupport {
private TestEntity entity;
+ private Location loc;
@BeforeMethod(alwaysRun=true)
@Override
public void setUp() throws Exception {
super.setUp();
entity = app.createAndManageChild(EntitySpec.create(TestEntity.class).displayName("mydisplayname"));
+ loc = app.getManagementContext().getLocationRegistry().resolve("localhost");
}
@Test
@@ -62,4 +68,10 @@ public class EntityFunctionsTest extends BrooklynAppUnitTestSupport {
public void testId() throws Exception {
assertEquals(EntityFunctions.id().apply(entity), entity.getId());
}
+
+ @Test
+ public void testLocationMatching() throws Exception {
+ entity.addLocations(ImmutableList.of(loc));
+ assertEquals(EntityFunctions.locationMatching(Predicates.alwaysTrue()).apply(entity), loc);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/159745bf/core/src/test/java/brooklyn/entity/basic/EntitySuppliersTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/entity/basic/EntitySuppliersTest.java b/core/src/test/java/brooklyn/entity/basic/EntitySuppliersTest.java
new file mode 100644
index 0000000..ba41199
--- /dev/null
+++ b/core/src/test/java/brooklyn/entity/basic/EntitySuppliersTest.java
@@ -0,0 +1,70 @@
+/*
+ * 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 brooklyn.entity.basic;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.fail;
+
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import brooklyn.entity.BrooklynAppUnitTestSupport;
+import brooklyn.entity.proxying.EntitySpec;
+import brooklyn.location.Location;
+import brooklyn.location.MachineProvisioningLocation;
+import brooklyn.location.basic.SshMachineLocation;
+import brooklyn.test.entity.TestEntity;
+
+import com.google.common.base.Supplier;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+
+public class EntitySuppliersTest extends BrooklynAppUnitTestSupport {
+
+ private TestEntity entity;
+ private Location loc;
+ private SshMachineLocation machine;
+
+ @BeforeMethod(alwaysRun=true)
+ @SuppressWarnings("unchecked")
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
+ entity = app.createAndManageChild(EntitySpec.create(TestEntity.class).displayName("mydisplayname"));
+ loc = app.getManagementContext().getLocationRegistry().resolve("localhost");
+ machine = ((MachineProvisioningLocation<SshMachineLocation>)loc).obtain(ImmutableMap.of());
+ }
+
+ @Test
+ public void testUniqueSshMachineLocation() throws Exception {
+ entity.addLocations(ImmutableList.of(machine));
+ assertEquals(EntitySuppliers.uniqueSshMachineLocation(entity).get(), machine);
+ }
+
+ @Test
+ public void testUniqueSshMachineLocationWhenNoLocation() throws Exception {
+ Supplier<SshMachineLocation> supplier = EntitySuppliers.uniqueSshMachineLocation(entity);
+ try {
+ supplier.get();
+ fail();
+ } catch (IllegalStateException e) {
+ // expected: success
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/159745bf/core/src/test/java/brooklyn/entity/rebind/RebindFeedTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/entity/rebind/RebindFeedTest.java b/core/src/test/java/brooklyn/entity/rebind/RebindFeedTest.java
index 744f948..b4e2a99 100644
--- a/core/src/test/java/brooklyn/entity/rebind/RebindFeedTest.java
+++ b/core/src/test/java/brooklyn/entity/rebind/RebindFeedTest.java
@@ -178,7 +178,6 @@ public class RebindFeedTest extends RebindTestFixtureWithApp {
public static class MyEntityWithSshFeedImpl extends TestEntityImpl {
@Override
public void start(Collection<? extends Location> locs) {
- // TODO Auto-generated method stub
super.start(locs);
addFeed(SshFeed.builder()
.entity(this)
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/159745bf/core/src/test/java/brooklyn/event/feed/windows/WindowsPerformanceCounterFeedLiveTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/event/feed/windows/WindowsPerformanceCounterFeedLiveTest.java b/core/src/test/java/brooklyn/event/feed/windows/WindowsPerformanceCounterFeedLiveTest.java
index 34ee43b..9a9210f 100644
--- a/core/src/test/java/brooklyn/event/feed/windows/WindowsPerformanceCounterFeedLiveTest.java
+++ b/core/src/test/java/brooklyn/event/feed/windows/WindowsPerformanceCounterFeedLiveTest.java
@@ -20,13 +20,10 @@ package brooklyn.event.feed.windows;
import java.util.Map;
-import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
-import brooklyn.entity.basic.ApplicationBuilder;
-import brooklyn.entity.basic.Entities;
-import brooklyn.entity.basic.EntityInternal;
+import brooklyn.entity.BrooklynAppLiveTestSupport;
import brooklyn.entity.basic.EntityLocal;
import brooklyn.entity.proxying.EntitySpec;
import brooklyn.event.AttributeSensor;
@@ -34,9 +31,7 @@ import brooklyn.event.basic.Sensors;
import brooklyn.location.Location;
import brooklyn.location.MachineLocation;
import brooklyn.location.MachineProvisioningLocation;
-import brooklyn.management.ManagementContext;
import brooklyn.test.EntityTestUtils;
-import brooklyn.test.entity.TestApplication;
import brooklyn.test.entity.TestEntity;
import brooklyn.util.collections.MutableMap;
@@ -64,29 +59,20 @@ import com.google.common.collect.ImmutableMap;
* use a jclouds location, as adding a dependency on brooklyn-locations-jclouds would cause a
* cyclic dependency.
*/
-public class WindowsPerformanceCounterFeedLiveTest {
+public class WindowsPerformanceCounterFeedLiveTest extends BrooklynAppLiveTestSupport {
- final static AttributeSensor<Double> CPU_IDLE_TIME =
- Sensors.newDoubleSensor("cpu.idleTime", "");
- final static AttributeSensor<Integer> TELEPHONE_LINES =
- Sensors.newIntegerSensor("telephone.lines", "");
+ final static AttributeSensor<Double> CPU_IDLE_TIME = Sensors.newDoubleSensor("cpu.idleTime", "");
+ final static AttributeSensor<Integer> TELEPHONE_LINES = Sensors.newIntegerSensor("telephone.lines", "");
private static final String LOCATION_SPEC = "named:WindowsLiveTest";
- private ManagementContext mgmt;
- private TestApplication app;
private Location loc;
private EntityLocal entity;
@BeforeMethod(alwaysRun=true)
public void setUp() throws Exception {
- if (mgmt!=null) {
- app = ApplicationBuilder.newManagedApp(TestApplication.class, mgmt);
- } else {
- app = ApplicationBuilder.newManagedApp(TestApplication.class);
- mgmt = ((EntityInternal)app).getManagementContext();
- }
-
+ super.setUp();
+
Map<String,?> allFlags = MutableMap.<String,Object>builder()
.put("tags", ImmutableList.of(getClass().getName()))
.build();
@@ -99,12 +85,6 @@ public class WindowsPerformanceCounterFeedLiveTest {
app.start(ImmutableList.of(loc));
}
- @AfterMethod(alwaysRun=true)
- public void tearDown() throws Exception {
- if (mgmt != null) Entities.destroyAllCatching(mgmt);
- mgmt = null;
- }
-
@Test(groups={"Live","Disabled"}, enabled=false)
public void testRetrievesPerformanceCounters() throws Exception {
// We can be pretty sure that a Windows instance in the cloud will have zero telephone lines...
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/159745bf/core/src/test/java/brooklyn/event/feed/windows/WindowsPerformanceCounterFeedTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/brooklyn/event/feed/windows/WindowsPerformanceCounterFeedTest.java b/core/src/test/java/brooklyn/event/feed/windows/WindowsPerformanceCounterFeedTest.java
index 61304ab..4b2ed91 100644
--- a/core/src/test/java/brooklyn/event/feed/windows/WindowsPerformanceCounterFeedTest.java
+++ b/core/src/test/java/brooklyn/event/feed/windows/WindowsPerformanceCounterFeedTest.java
@@ -32,7 +32,7 @@ public class WindowsPerformanceCounterFeedTest {
@Test
public void testIteratorWithSingleValue() {
- Iterator iterator = new WindowsPerformanceCounterFeed
+ Iterator<?> iterator = new WindowsPerformanceCounterFeed
.PerfCounterValueIterator("\"10/14/2013 15:28:24.406\",\"0.000000\"");
assertTrue(iterator.hasNext());
assertEquals(iterator.next(), "0.000000");
@@ -41,7 +41,7 @@ public class WindowsPerformanceCounterFeedTest {
@Test
public void testIteratorWithMultipleValues() {
- Iterator iterator = new WindowsPerformanceCounterFeed
+ Iterator<?> iterator = new WindowsPerformanceCounterFeed
.PerfCounterValueIterator("\"10/14/2013 15:35:50.582\",\"8803.000000\",\"405622.000000\"");
assertTrue(iterator.hasNext());
assertEquals(iterator.next(), "8803.000000");
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/159745bf/sandbox/monitoring/src/main/java/brooklyn/entity/monitoring/zabbix/ZabbixFeed.java
----------------------------------------------------------------------
diff --git a/sandbox/monitoring/src/main/java/brooklyn/entity/monitoring/zabbix/ZabbixFeed.java b/sandbox/monitoring/src/main/java/brooklyn/entity/monitoring/zabbix/ZabbixFeed.java
index 55645f4..e1b022d 100644
--- a/sandbox/monitoring/src/main/java/brooklyn/entity/monitoring/zabbix/ZabbixFeed.java
+++ b/sandbox/monitoring/src/main/java/brooklyn/entity/monitoring/zabbix/ZabbixFeed.java
@@ -35,8 +35,11 @@ import org.apache.http.impl.conn.tsccm.ThreadSafeClientConnManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import brooklyn.config.ConfigKey;
import brooklyn.entity.Entity;
import brooklyn.entity.basic.Attributes;
+import brooklyn.entity.basic.ConfigKeys;
+import brooklyn.entity.basic.EntityFunctions;
import brooklyn.entity.basic.EntityLocal;
import brooklyn.event.feed.AbstractFeed;
import brooklyn.event.feed.AttributePollHandler;
@@ -52,6 +55,7 @@ import brooklyn.util.http.HttpToolResponse;
import brooklyn.util.net.Cidr;
import com.google.common.base.Function;
+import com.google.common.base.Functions;
import com.google.common.base.Objects;
import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
@@ -63,6 +67,7 @@ import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.google.common.net.HostAndPort;
+import com.google.common.reflect.TypeToken;
import com.google.gson.JsonObject;
public class ZabbixFeed extends AbstractFeed {
@@ -88,6 +93,25 @@ public class ZabbixFeed extends AbstractFeed {
private static final AtomicInteger id = new AtomicInteger(0);
+ @SuppressWarnings("serial")
+ public static final ConfigKey<Set<ZabbixPollConfig<?>>> POLLS = ConfigKeys.newConfigKey(
+ new TypeToken<Set<ZabbixPollConfig<?>>>() {},
+ "polls");
+
+ @SuppressWarnings("serial")
+ public static final ConfigKey<Supplier<URI>> BASE_URI_PROVIDER = ConfigKeys.newConfigKey(
+ new TypeToken<Supplier<URI>>() {},
+ "baseUriProvider");
+
+ public static final ConfigKey<Integer> GROUP_ID = ConfigKeys.newIntegerConfigKey("groupId");
+
+ public static final ConfigKey<Integer> TEMPLATE_ID = ConfigKeys.newIntegerConfigKey("templateId");
+
+ @SuppressWarnings("serial")
+ public static final ConfigKey<Function<? super EntityLocal, String>> UNIQUE_HOSTNAME_GENERATOR = ConfigKeys.newConfigKey(
+ new TypeToken<Function<? super EntityLocal, String>>() {},
+ "uniqueHostnameGenerator");
+
public static Builder<ZabbixFeed, ?> builder() {
return new ConcreteBuilder();
}
@@ -110,11 +134,9 @@ public class ZabbixFeed extends AbstractFeed {
private Integer sessionTimeout;
private Integer groupId;
private Integer templateId;
- private Function<? super EntityLocal, String> uniqueHostnameGenerator = new Function<EntityLocal, String>() {
- @Override public String apply(EntityLocal entity) {
- Location loc = Iterables.find(entity.getLocations(), Predicates.instanceOf(MachineLocation.class));
- return loc.getId();
- }};
+ private Function<? super EntityLocal, String> uniqueHostnameGenerator = Functions.compose(
+ EntityFunctions.id(),
+ EntityFunctions.locationMatching(Predicates.instanceOf(MachineLocation.class)));
@SuppressWarnings("unchecked")
protected B self() {
@@ -217,6 +239,7 @@ public class ZabbixFeed extends AbstractFeed {
}
// Now create feed
T result = (T) new ZabbixFeed(this);
+ result.setEntity(checkNotNull(entity, "entity"));
built = true;
if (suspended) result.suspend();
result.start();
@@ -250,44 +273,48 @@ public class ZabbixFeed extends AbstractFeed {
}
}
- // Treat as immutable once built
- protected final Set<ZabbixPollConfig<?>> polls = Sets.newLinkedHashSet();
-
- protected Supplier<URI> baseUriProvider;
- protected Integer groupId, templateId;
-
// Flag set when the Zabbix agent is registered for a host
protected final AtomicBoolean registered = new AtomicBoolean(false);
- private final Function<? super EntityLocal, String> uniqueHostnameGenerator;
-
+ /**
+ * For rebind; do not call directly; use builder
+ */
+ public ZabbixFeed() {
+ }
+
protected ZabbixFeed(final Builder<? extends ZabbixFeed, ?> builder) {
- super(builder.entity);
-
- baseUriProvider = builder.baseUriProvider;
- if (builder.baseUri!=null) {
- if (baseUriProvider!=null)
+ setConfig(BASE_URI_PROVIDER, builder.baseUriProvider);
+ if (builder.baseUri != null) {
+ if (builder.baseUriProvider != null) {
throw new IllegalStateException("Not permitted to supply baseUri and baseUriProvider");
- URI uri = builder.baseUri;
- baseUriProvider = Suppliers.ofInstance(uri);
+ }
+ setConfig(BASE_URI_PROVIDER, Suppliers.ofInstance(builder.baseUri));
+ } else {
+ setConfig(BASE_URI_PROVIDER, checkNotNull(builder.baseUriProvider, "baseUriProvider and baseUri"));
}
- checkNotNull(baseUriProvider);
- groupId = checkNotNull(builder.groupId, "Zabbix groupId must be set");
- templateId = checkNotNull(builder.templateId, "Zabbix templateId must be set");
+ setConfig(GROUP_ID, checkNotNull(builder.groupId, "Zabbix groupId must be set"));
+ setConfig(TEMPLATE_ID, checkNotNull(builder.templateId, "Zabbix templateId must be set"));
+ setConfig(UNIQUE_HOSTNAME_GENERATOR, checkNotNull(builder.uniqueHostnameGenerator, "uniqueHostnameGenerator"));
+ Set<ZabbixPollConfig<?>> polls = Sets.newLinkedHashSet();
for (ZabbixPollConfig<?> config : builder.polls) {
@SuppressWarnings({ "unchecked", "rawtypes" })
ZabbixPollConfig<?> configCopy = new ZabbixPollConfig(config);
if (configCopy.getPeriod() < 0) configCopy.period(builder.period, builder.periodUnits);
polls.add(configCopy);
}
-
- uniqueHostnameGenerator = checkNotNull(builder.uniqueHostnameGenerator, "uniqueHostnameGenerator");
+ setConfig(POLLS, polls);
}
@Override
protected void preStart() {
+ final Supplier<URI> baseUriProvider = getConfig(BASE_URI_PROVIDER);
+ final Function<? super EntityLocal, String> uniqueHostnameGenerator = getConfig(UNIQUE_HOSTNAME_GENERATOR);
+ final Integer groupId = getConfig(GROUP_ID);
+ final Integer templateId = getConfig(TEMPLATE_ID);
+ final Set<ZabbixPollConfig<?>> polls = getConfig(POLLS);
+
log.info("starting zabbix feed for {}", entity);
// TODO if supplier returns null, we may wish to defer initialization until url available?
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/159745bf/software/base/src/main/java/brooklyn/entity/chef/ChefAttributeFeed.java
----------------------------------------------------------------------
diff --git a/software/base/src/main/java/brooklyn/entity/chef/ChefAttributeFeed.java b/software/base/src/main/java/brooklyn/entity/chef/ChefAttributeFeed.java
index f842db2..e27b391 100644
--- a/software/base/src/main/java/brooklyn/entity/chef/ChefAttributeFeed.java
+++ b/software/base/src/main/java/brooklyn/entity/chef/ChefAttributeFeed.java
@@ -24,12 +24,16 @@ import static com.google.common.base.Preconditions.checkNotNull;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.SortedSet;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import brooklyn.config.ConfigKey;
+import brooklyn.entity.basic.ConfigKeys;
import brooklyn.entity.basic.EntityInternal;
import brooklyn.entity.basic.EntityLocal;
import brooklyn.event.AttributeSensor;
@@ -50,6 +54,8 @@ import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import com.google.common.reflect.TypeToken;
import com.google.gson.Gson;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
@@ -87,13 +93,20 @@ import com.google.gson.JsonObject;
*/
public class ChefAttributeFeed extends AbstractFeed {
+ private static final Logger log = LoggerFactory.getLogger(ChefAttributeFeed.class);
+
/**
* Prefix for attribute sensor names.
*/
public static final String CHEF_ATTRIBUTE_PREFIX = "chef.attribute.";
- private static final Logger log = LoggerFactory.getLogger(ChefAttributeFeed.class);
+ @SuppressWarnings("serial")
+ public static final ConfigKey<Set<ChefAttributePollConfig<?>>> POLLS = ConfigKeys.newConfigKey(
+ new TypeToken<Set<ChefAttributePollConfig<?>>>() {},
+ "polls");
+ public static final ConfigKey<String> NODE_NAME = ConfigKeys.newStringConfigKey("nodeName");
+
public static Builder builder() {
return new Builder();
}
@@ -103,9 +116,8 @@ public class ChefAttributeFeed extends AbstractFeed {
private EntityLocal entity;
private boolean onlyIfServiceUp = false;
private String nodeName;
- private Map<String, AttributeSensor<?>> sensors = Maps.newHashMap();
- private long period = 30;
- private TimeUnit periodUnits = TimeUnit.SECONDS;
+ private Set<ChefAttributePollConfig> polls = Sets.newLinkedHashSet();
+ private Duration period = Duration.of(30, TimeUnit.SECONDS);
private volatile boolean built;
public Builder entity(EntityLocal val) {
@@ -121,12 +133,17 @@ public class ChefAttributeFeed extends AbstractFeed {
this.nodeName = checkNotNull(nodeName, "nodeName");
return this;
}
- public Builder addSensor(String chefAttributePath, AttributeSensor sensor) {
- sensors.put(checkNotNull(chefAttributePath, "chefAttributePath"), checkNotNull(sensor, "sensor"));
+ public Builder addSensor(ChefAttributePollConfig config) {
+ polls.add(config);
return this;
}
+ public Builder addSensor(String chefAttributePath, AttributeSensor sensor) {
+ return addSensor(new ChefAttributePollConfig(sensor).chefAttributePath(chefAttributePath));
+ }
public Builder addSensors(Map<String, AttributeSensor> sensors) {
- sensors.putAll(checkNotNull(sensors, "sensors"));
+ for (Map.Entry<String, AttributeSensor> entry : sensors.entrySet()) {
+ addSensor(entry.getKey(), entry.getValue());
+ }
return this;
}
public Builder addSensors(AttributeSensor[] sensors) {
@@ -141,19 +158,19 @@ public class ChefAttributeFeed extends AbstractFeed {
return this;
}
public Builder period(Duration period) {
- return period(checkNotNull(period, "period").toMilliseconds(), TimeUnit.MILLISECONDS);
+ this.period = period;
+ return this;
}
public Builder period(long millis) {
- return period(millis, TimeUnit.MILLISECONDS);
+ return period(Duration.of(millis, TimeUnit.MILLISECONDS));
}
public Builder period(long val, TimeUnit units) {
- this.period = val;
- this.periodUnits = checkNotNull(units, "units");
- return this;
+ return period(Duration.of(val, units));
}
public ChefAttributeFeed build() {
built = true;
ChefAttributeFeed result = new ChefAttributeFeed(this);
+ result.setEntity(checkNotNull(entity, "entity"));
result.start();
return result;
}
@@ -163,23 +180,41 @@ public class ChefAttributeFeed extends AbstractFeed {
}
}
- private final String nodeName;
- private final long period;
- private final TimeUnit periodUnits;
- private final Map<String, AttributeSensor<?>> chefAttributeSensors;
- private final KnifeTaskFactory<String> knifeTaskFactory;
+ private KnifeTaskFactory<String> knifeTaskFactory;
+ /**
+ * For rebind; do not call directly; use builder
+ */
+ public ChefAttributeFeed() {
+ }
+
protected ChefAttributeFeed(Builder builder) {
- super(checkNotNull(builder.entity, "entity"), builder.onlyIfServiceUp);
- nodeName = checkNotNull(builder.nodeName, "builder.nodeName");
- period = builder.period;
- periodUnits = builder.periodUnits;
- chefAttributeSensors = builder.sensors;
- knifeTaskFactory = new KnifeNodeAttributeQueryTaskFactory(nodeName);
+ setConfig(ONLY_IF_SERVICE_UP, builder.onlyIfServiceUp);
+ setConfig(NODE_NAME, checkNotNull(builder.nodeName, "builder.nodeName"));
+
+ Set<ChefAttributePollConfig<?>> polls = Sets.newLinkedHashSet();
+ for (ChefAttributePollConfig<?> config : builder.polls) {
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ ChefAttributePollConfig<?> configCopy = new ChefAttributePollConfig(config);
+ if (configCopy.getPeriod() < 0) configCopy.period(builder.period);
+ polls.add(configCopy);
+ }
+ setConfig(POLLS, polls);
}
@Override
protected void preStart() {
+ final String nodeName = getConfig(NODE_NAME);
+ final Set<ChefAttributePollConfig<?>> polls = getConfig(POLLS);
+
+ long minPeriod = Integer.MAX_VALUE;
+ SortedSet<String> performanceCounterNames = Sets.newTreeSet();
+ for (ChefAttributePollConfig<?> config : polls) {
+ minPeriod = Math.min(minPeriod, config.getPeriod());
+ }
+
+ knifeTaskFactory = new KnifeNodeAttributeQueryTaskFactory(nodeName);
+
final Callable<SshPollValue> getAttributesFromKnife = new Callable<SshPollValue>() {
public SshPollValue call() throws Exception {
ProcessTaskWrapper<String> taskWrapper = knifeTaskFactory.newTask();
@@ -194,8 +229,8 @@ public class ChefAttributeFeed extends AbstractFeed {
getPoller().scheduleAtFixedRate(
new CallInEntityExecutionContext<SshPollValue>(entity, getAttributesFromKnife),
- new SendChefAttributesToSensors(entity, chefAttributeSensors),
- periodUnits.toMillis(period));
+ new SendChefAttributesToSensors(entity, polls),
+ minPeriod);
}
@SuppressWarnings("unchecked")
@@ -250,12 +285,15 @@ public class ChefAttributeFeed extends AbstractFeed {
private static final Iterable<String> PREFIXES = ImmutableList.of("", "automatic", "force_override", "override", "normal", "force_default", "default");
private static final Splitter SPLITTER = Splitter.on('.');
- private Map<String, AttributeSensor<?>> chefAttributeSensors;
- private EntityLocal entity;
+ private final EntityLocal entity;
+ private final Map<String, AttributeSensor<?>> chefAttributeSensors;
- public SendChefAttributesToSensors(EntityLocal entity, Map<String, AttributeSensor<?>> chefAttributeSensors) {
- this.chefAttributeSensors = chefAttributeSensors;
+ public SendChefAttributesToSensors(EntityLocal entity, Set<ChefAttributePollConfig<?>> polls) {
this.entity = entity;
+ chefAttributeSensors = Maps.newLinkedHashMap();
+ for (ChefAttributePollConfig<?> config : polls) {
+ chefAttributeSensors.put(config.getChefAttributePath(), config.getSensor());
+ }
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/159745bf/software/base/src/main/java/brooklyn/entity/chef/ChefAttributePollConfig.java
----------------------------------------------------------------------
diff --git a/software/base/src/main/java/brooklyn/entity/chef/ChefAttributePollConfig.java b/software/base/src/main/java/brooklyn/entity/chef/ChefAttributePollConfig.java
new file mode 100644
index 0000000..b76b6ad
--- /dev/null
+++ b/software/base/src/main/java/brooklyn/entity/chef/ChefAttributePollConfig.java
@@ -0,0 +1,54 @@
+/*
+ * 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 brooklyn.entity.chef;
+
+import brooklyn.event.AttributeSensor;
+import brooklyn.event.feed.PollConfig;
+
+import com.google.common.base.Function;
+import com.google.common.base.Functions;
+
+public class ChefAttributePollConfig<T> extends PollConfig<Object, T, ChefAttributePollConfig<T>>{
+
+ private String chefAttributePath;
+
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ public ChefAttributePollConfig(AttributeSensor<T> sensor) {
+ super(sensor);
+ onSuccess((Function)Functions.identity());
+ }
+
+ public ChefAttributePollConfig(ChefAttributePollConfig<T> other) {
+ super(other);
+ this.chefAttributePath = other.chefAttributePath;
+ }
+
+ public String getChefAttributePath() {
+ return chefAttributePath;
+ }
+
+ public ChefAttributePollConfig<T> chefAttributePath(String val) {
+ this.chefAttributePath = val; return this;
+ }
+
+ @Override
+ public String toString() {
+ return "chef["+chefAttributePath+"]";
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/159745bf/software/base/src/main/java/brooklyn/event/feed/jmx/JmxFeed.java
----------------------------------------------------------------------
diff --git a/software/base/src/main/java/brooklyn/event/feed/jmx/JmxFeed.java b/software/base/src/main/java/brooklyn/event/feed/jmx/JmxFeed.java
index ab05914..6ea4933 100644
--- a/software/base/src/main/java/brooklyn/event/feed/jmx/JmxFeed.java
+++ b/software/base/src/main/java/brooklyn/event/feed/jmx/JmxFeed.java
@@ -18,6 +18,8 @@
*/
package brooklyn.event.feed.jmx;
+import static com.google.common.base.Preconditions.checkNotNull;
+
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -32,6 +34,8 @@ import javax.management.ObjectName;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import brooklyn.config.ConfigKey;
+import brooklyn.entity.basic.ConfigKeys;
import brooklyn.entity.basic.EntityLocal;
import brooklyn.entity.basic.SoftwareProcessImpl;
import brooklyn.event.feed.AbstractFeed;
@@ -41,12 +45,12 @@ import brooklyn.event.feed.PollHandler;
import brooklyn.event.feed.Poller;
import brooklyn.util.time.Duration;
-import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.SetMultimap;
import com.google.common.collect.Sets;
+import com.google.common.reflect.TypeToken;
/**
@@ -90,6 +94,26 @@ public class JmxFeed extends AbstractFeed {
public static final long JMX_CONNECTION_TIMEOUT_MS = 120*1000;
+ public static final ConfigKey<JmxHelper> HELPER = ConfigKeys.newConfigKey(JmxHelper.class, "helper");
+ public static final ConfigKey<Boolean> OWN_HELPER = ConfigKeys.newBooleanConfigKey("ownHelper");
+ public static final ConfigKey<String> JMX_URI = ConfigKeys.newStringConfigKey("jmxUri");
+ public static final ConfigKey<Long> JMX_CONNECTION_TIMEOUT = ConfigKeys.newLongConfigKey("jmxConnectionTimeout");
+
+ @SuppressWarnings("serial")
+ public static final ConfigKey<SetMultimap<String, JmxAttributePollConfig<?>>> ATTRIBUTE_POLLS = ConfigKeys.newConfigKey(
+ new TypeToken<SetMultimap<String, JmxAttributePollConfig<?>>>() {},
+ "attributePolls");
+
+ @SuppressWarnings("serial")
+ public static final ConfigKey<SetMultimap<List<?>, JmxOperationPollConfig<?>>> OPERATION_POLLS = ConfigKeys.newConfigKey(
+ new TypeToken<SetMultimap<List<?>, JmxOperationPollConfig<?>>>() {},
+ "operationPolls");
+
+ @SuppressWarnings("serial")
+ public static final ConfigKey<SetMultimap<NotificationFilter, JmxNotificationSubscriptionConfig<?>>> NOTIFICATION_SUBSCRIPTIONS = ConfigKeys.newConfigKey(
+ new TypeToken<SetMultimap<NotificationFilter, JmxNotificationSubscriptionConfig<?>>>() {},
+ "notificationPolls");
+
public static Builder builder() {
return new Builder();
}
@@ -139,6 +163,7 @@ public class JmxFeed extends AbstractFeed {
public JmxFeed build() {
built = true;
JmxFeed result = new JmxFeed(this);
+ result.setEntity(checkNotNull(entity, "entity"));
result.start();
return result;
}
@@ -148,43 +173,66 @@ public class JmxFeed extends AbstractFeed {
}
}
- private final JmxHelper helper;
- private final boolean ownHelper;
- private final String jmxUri;
- private final long jmxConnectionTimeout;
-
- // Treat as immutable; never modified after constructor
- private final SetMultimap<String, JmxAttributePollConfig<?>> attributePolls = HashMultimap.<String,JmxAttributePollConfig<?>>create();
- private final SetMultimap<List<?>, JmxOperationPollConfig<?>> operationPolls = HashMultimap.<List<?>,JmxOperationPollConfig<?>>create();
- private final SetMultimap<NotificationFilter, JmxNotificationSubscriptionConfig<?>> notificationSubscriptions = HashMultimap.create();
private final SetMultimap<ObjectName, NotificationListener> notificationListeners = HashMultimap.create();
+ /**
+ * For rebind; do not call directly; use builder
+ */
+ public JmxFeed() {
+ }
+
protected JmxFeed(Builder builder) {
- super(builder.entity);
- this.helper = (builder.helper != null) ? builder.helper : new JmxHelper(entity);
- this.ownHelper = (builder.helper == null);
- this.jmxUri = helper.getUrl();
- this.jmxConnectionTimeout = builder.jmxConnectionTimeout;
+ super();
+ if (builder.helper != null) {
+ JmxHelper helper = builder.helper;
+ setConfig(HELPER, helper);
+ setConfig(OWN_HELPER, false);
+ setConfig(JMX_URI, helper.getUrl());
+ }
+ setConfig(JMX_CONNECTION_TIMEOUT, builder.jmxConnectionTimeout);
+ SetMultimap<String, JmxAttributePollConfig<?>> attributePolls = HashMultimap.<String,JmxAttributePollConfig<?>>create();
for (JmxAttributePollConfig<?> config : builder.attributePolls) {
@SuppressWarnings({ "rawtypes", "unchecked" })
JmxAttributePollConfig<?> configCopy = new JmxAttributePollConfig(config);
if (configCopy.getPeriod() < 0) configCopy.period(builder.period, builder.periodUnits);
attributePolls.put(configCopy.getObjectName().getCanonicalName() + configCopy.getAttributeName(), configCopy);
}
+ setConfig(ATTRIBUTE_POLLS, attributePolls);
+
+ SetMultimap<List<?>, JmxOperationPollConfig<?>> operationPolls = HashMultimap.<List<?>,JmxOperationPollConfig<?>>create();
for (JmxOperationPollConfig<?> config : builder.operationPolls) {
@SuppressWarnings({ "rawtypes", "unchecked" })
JmxOperationPollConfig<?> configCopy = new JmxOperationPollConfig(config);
if (configCopy.getPeriod() < 0) configCopy.period(builder.period, builder.periodUnits);
operationPolls.put(configCopy.buildOperationIdentity(), configCopy);
}
+ setConfig(OPERATION_POLLS, operationPolls);
+
+ SetMultimap<NotificationFilter, JmxNotificationSubscriptionConfig<?>> notificationSubscriptions = HashMultimap.create();
for (JmxNotificationSubscriptionConfig<?> config : builder.notificationSubscriptions) {
notificationSubscriptions.put(config.getNotificationFilter(), config);
}
+ setConfig(NOTIFICATION_SUBSCRIPTIONS, notificationSubscriptions);
}
+ @Override
+ public void setEntity(EntityLocal entity) {
+ if (getConfig(HELPER) == null) {
+ JmxHelper helper = new JmxHelper(entity);
+ setConfig(HELPER, helper);
+ setConfig(OWN_HELPER, true);
+ setConfig(JMX_URI, helper.getUrl());
+ }
+ super.setEntity(entity);
+ }
+
public String getJmxUri() {
- return jmxUri;
+ return getConfig(JMX_URI);
+ }
+
+ protected JmxHelper getHelper() {
+ return getConfig(HELPER);
}
@SuppressWarnings("unchecked")
@@ -194,7 +242,7 @@ public class JmxFeed extends AbstractFeed {
@Override
protected boolean isConnected() {
- return super.isConnected() && helper.isConnected();
+ return super.isConnected() && getHelper().isConnected();
}
@Override
@@ -217,13 +265,16 @@ public class JmxFeed extends AbstractFeed {
* ...
* at brooklyn.entity.rebind.RebindManagerImpl.rebind(RebindManagerImpl.java:184)
*/
+ final SetMultimap<NotificationFilter, JmxNotificationSubscriptionConfig<?>> notificationSubscriptions = getConfig(NOTIFICATION_SUBSCRIPTIONS);
+ final SetMultimap<List<?>, JmxOperationPollConfig<?>> operationPolls = getConfig(OPERATION_POLLS);
+ final SetMultimap<String, JmxAttributePollConfig<?>> attributePolls = getConfig(ATTRIBUTE_POLLS);
getPoller().submit(new Callable<Void>() {
public Void call() {
- helper.connect(jmxConnectionTimeout);
+ getHelper().connect(getConfig(JMX_CONNECTION_TIMEOUT));
return null;
}
- @Override public String toString() { return "Connect JMX "+helper.getUrl(); }
+ @Override public String toString() { return "Connect JMX "+getHelper().getUrl(); }
});
for (final NotificationFilter filter : notificationSubscriptions.keySet()) {
@@ -254,7 +305,7 @@ public class JmxFeed extends AbstractFeed {
@Override
protected void preStop() {
super.preStop();
-
+
for (Map.Entry<ObjectName, NotificationListener> entry : notificationListeners.entries()) {
unregisterNotificationListener(entry.getKey(), entry.getValue());
}
@@ -264,6 +315,8 @@ public class JmxFeed extends AbstractFeed {
@Override
protected void postStop() {
super.postStop();
+ JmxHelper helper = getHelper();
+ Boolean ownHelper = getConfig(OWN_HELPER);
if (helper != null && ownHelper) helper.terminate();
}
@@ -287,11 +340,11 @@ public class JmxFeed extends AbstractFeed {
getPoller().scheduleAtFixedRate(
new Callable<Object>() {
public Object call() throws Exception {
- if (log.isDebugEnabled()) log.debug("jmx operation polling for {} sensors at {} -> {}", new Object[] {getEntity(), jmxUri, operationName});
+ if (log.isDebugEnabled()) log.debug("jmx operation polling for {} sensors at {} -> {}", new Object[] {getEntity(), getJmxUri(), operationName});
if (signature.size() == params.size()) {
- return helper.operation(objectName, operationName, signature, params);
+ return getHelper().operation(objectName, operationName, signature, params);
} else {
- return helper.operation(objectName, operationName, params.toArray());
+ return getHelper().operation(objectName, operationName, params.toArray());
}
}
},
@@ -317,8 +370,8 @@ public class JmxFeed extends AbstractFeed {
getPoller().scheduleAtFixedRate(
new Callable<Object>() {
public Object call() throws Exception {
- if (log.isTraceEnabled()) log.trace("jmx attribute polling for {} sensors at {} -> {}", new Object[] {getEntity(), jmxUri, jmxAttributeName});
- return helper.getAttribute(objectName, jmxAttributeName);
+ if (log.isTraceEnabled()) log.trace("jmx attribute polling for {} sensors at {} -> {}", new Object[] {getEntity(), getJmxUri(), jmxAttributeName});
+ return getHelper().getAttribute(objectName, jmxAttributeName);
}
},
new DelegatingPollHandler<Object>(handlers), minPeriod);
@@ -355,14 +408,14 @@ public class JmxFeed extends AbstractFeed {
compoundHandler.onSuccess(notification);
}
};
- helper.addNotificationListener(objectName, listener, filter);
+ getHelper().addNotificationListener(objectName, listener, filter);
return listener;
}
private void unregisterNotificationListener(ObjectName objectName, NotificationListener listener) {
try {
- helper.removeNotificationListener(objectName, listener);
+ getHelper().removeNotificationListener(objectName, listener);
} catch (RuntimeException e) {
log.warn("Failed to unregister listener: "+objectName+", "+listener+"; continuing...", e);
}
@@ -370,6 +423,6 @@ public class JmxFeed extends AbstractFeed {
@Override
public String toString() {
- return "JmxFeed["+jmxUri+"]";
+ return "JmxFeed["+getJmxUri()+"]";
}
}