diff --git oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation/filter/FilterBuilder.java oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation/filter/FilterBuilder.java index 1a3a89f..cb76793 100644 --- oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation/filter/FilterBuilder.java +++ oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation/filter/FilterBuilder.java @@ -39,6 +39,7 @@ import javax.annotation.Nonnull; import com.google.common.base.Objects; import com.google.common.base.Predicate; import com.google.common.collect.ImmutableList; +import com.google.common.collect.Iterables; import org.apache.jackrabbit.oak.api.PropertyState; import org.apache.jackrabbit.oak.plugins.nodetype.TypePredicate; import org.apache.jackrabbit.oak.plugins.observation.filter.UniversalFilter.Selector; @@ -391,6 +392,11 @@ public final class FilterBuilder { return subTrees; } + @Override + public FilterConfigMBean getConfigMBean() { + return FilterBuilder.this.getConfigMBean(); + } + private boolean isLocal(String sessionId, CommitInfo info) { return info != null && Objects.equal(info.getSessionId(), sessionId); } @@ -401,6 +407,26 @@ public final class FilterBuilder { }; } + @Nonnull + private FilterConfigMBean getConfigMBean(){ + return new FilterConfigMBean() { + @Override + public String[] getSubTrees() { + return Iterables.toArray(subTrees, String.class); + } + + @Override + public boolean isIncludeClusterLocal() { + return FilterBuilder.this.includeClusterExternal; + } + + @Override + public boolean isIncludeClusterExternal() { + return FilterBuilder.this.includeClusterLocal; + } + }; + } + //------------------------------------------------------------< Conditions >--- private static class ConstantCondition implements Condition { diff --git oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation/filter/FilterConfigMBean.java oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation/filter/FilterConfigMBean.java new file mode 100644 index 0000000..2a44a90 --- /dev/null +++ oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation/filter/FilterConfigMBean.java @@ -0,0 +1,45 @@ +/* + * 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.jackrabbit.oak.plugins.observation.filter; + +public interface FilterConfigMBean { + String TYPE = "FilterConfig"; + + /** + * A set of paths whose subtrees include all events of this filter. + * @return list of paths + * @see org.apache.jackrabbit.oak.plugins.observation.filter.FilterBuilder#addSubTree(String) + */ + String[] getSubTrees(); + + /** + * Whether to include cluster local changes. + * + * @return true if cluster local changes need to be included + */ + boolean isIncludeClusterLocal(); + + /** + * Whether to include cluster external changes. + * + * @return true if cluster external changes need to be included + */ + boolean isIncludeClusterExternal(); +} diff --git oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation/filter/FilterProvider.java oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation/filter/FilterProvider.java index 8dcdb11..779afed 100644 --- oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation/filter/FilterProvider.java +++ oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/observation/filter/FilterProvider.java @@ -62,4 +62,6 @@ public interface FilterProvider { */ @Nonnull Iterable getSubTrees(); + + FilterConfigMBean getConfigMBean(); } diff --git oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/BackgroundObserver.java oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/BackgroundObserver.java index 4dffa92..8d5762c 100644 --- oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/BackgroundObserver.java +++ oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/BackgroundObserver.java @@ -21,6 +21,8 @@ package org.apache.jackrabbit.oak.spi.commit; import static com.google.common.base.Preconditions.checkNotNull; import static com.google.common.base.Preconditions.checkState; +import static com.google.common.collect.Iterables.filter; +import static com.google.common.collect.Iterables.size; import static com.google.common.collect.Queues.newArrayBlockingQueue; import java.io.Closeable; @@ -34,6 +36,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import javax.annotation.Nonnull; import javax.annotation.Nullable; +import com.google.common.base.Predicate; import org.apache.jackrabbit.oak.spi.state.NodeState; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -202,6 +205,46 @@ public class BackgroundObserver implements Observer, Closeable { stopped = true; } + @Nonnull + public BackgroundObserverMBean getMBean(){ + return new BackgroundObserverMBean() { + @Override + public String getClassName() { + return observer.getClass().getName(); + } + + @Override + public int getQueueSize() { + return queue.size(); + } + + @Override + public int getMaxQueueSize() { + return getMaxQueueLength(); + } + + @Override + public int getLocalEventCount() { + return size(filter(queue, new Predicate() { + @Override + public boolean apply(@Nullable ContentChange input) { + return input.info != null; + } + })); + } + + @Override + public int getExternalEventCount() { + return size(filter(queue, new Predicate() { + @Override + public boolean apply(@Nullable ContentChange input) { + return input.info == null; + } + })); + } + }; + } + //----------------------------------------------------------< Observer >-- /** diff --git oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/BackgroundObserverMBean.java oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/BackgroundObserverMBean.java new file mode 100644 index 0000000..c1f6eab --- /dev/null +++ oak-core/src/main/java/org/apache/jackrabbit/oak/spi/commit/BackgroundObserverMBean.java @@ -0,0 +1,34 @@ +/* + * 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.jackrabbit.oak.spi.commit; + +public interface BackgroundObserverMBean { + String TYPE = "BackgroundObserverStats"; + + int getQueueSize(); + + int getMaxQueueSize(); + + int getLocalEventCount(); + + int getExternalEventCount(); + + String getClassName(); +} diff --git oak-core/src/main/java/org/apache/jackrabbit/oak/spi/whiteboard/WhiteboardUtils.java oak-core/src/main/java/org/apache/jackrabbit/oak/spi/whiteboard/WhiteboardUtils.java index 525fc0d..b46328a 100644 --- oak-core/src/main/java/org/apache/jackrabbit/oak/spi/whiteboard/WhiteboardUtils.java +++ oak-core/src/main/java/org/apache/jackrabbit/oak/spi/whiteboard/WhiteboardUtils.java @@ -19,8 +19,10 @@ package org.apache.jackrabbit.oak.spi.whiteboard; import static com.google.common.base.Preconditions.checkNotNull; import static java.util.Collections.emptyMap; +import java.util.Collections; import java.util.Hashtable; import java.util.List; +import java.util.Map; import java.util.concurrent.atomic.AtomicLong; import javax.annotation.CheckForNull; @@ -61,8 +63,15 @@ public class WhiteboardUtils { public static Registration registerMBean( Whiteboard whiteboard, Class iface, T bean, String type, String name) { + return registerMBean(whiteboard, iface, bean, type, name, Collections.emptyMap()); + } + + public static Registration registerMBean( + Whiteboard whiteboard, + Class iface, T bean, String type, String name, Map attrs) { try { - Hashtable table = new Hashtable(); + + Hashtable table = new Hashtable(attrs); table.put("type", ObjectName.quote(type)); table.put("name", ObjectName.quote(name)); table.put("id", String.valueOf(COUNTER.incrementAndGet())); diff --git oak-jcr/src/main/java/org/apache/jackrabbit/oak/jcr/observation/ChangeProcessor.java oak-jcr/src/main/java/org/apache/jackrabbit/oak/jcr/observation/ChangeProcessor.java index b24c783..a5eda78 100644 --- oak-jcr/src/main/java/org/apache/jackrabbit/oak/jcr/observation/ChangeProcessor.java +++ oak-jcr/src/main/java/org/apache/jackrabbit/oak/jcr/observation/ChangeProcessor.java @@ -26,7 +26,9 @@ import static org.apache.jackrabbit.oak.spi.whiteboard.WhiteboardUtils.registerM import static org.apache.jackrabbit.oak.spi.whiteboard.WhiteboardUtils.registerObserver; import static org.apache.jackrabbit.oak.spi.whiteboard.WhiteboardUtils.scheduleWithFixedDelay; +import java.util.Map; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; @@ -36,6 +38,7 @@ import javax.jcr.observation.Event; import javax.jcr.observation.EventIterator; import javax.jcr.observation.EventListener; +import com.google.common.collect.ImmutableMap; import com.google.common.util.concurrent.Monitor; import com.google.common.util.concurrent.Monitor.Guard; import org.apache.jackrabbit.api.jmx.EventListenerMBean; @@ -44,9 +47,11 @@ import org.apache.jackrabbit.oak.api.ContentSession; import org.apache.jackrabbit.oak.namepath.NamePathMapper; import org.apache.jackrabbit.oak.plugins.observation.CommitRateLimiter; import org.apache.jackrabbit.oak.plugins.observation.filter.EventFilter; +import org.apache.jackrabbit.oak.plugins.observation.filter.FilterConfigMBean; import org.apache.jackrabbit.oak.plugins.observation.filter.FilterProvider; import org.apache.jackrabbit.oak.plugins.observation.filter.Filters; import org.apache.jackrabbit.oak.spi.commit.BackgroundObserver; +import org.apache.jackrabbit.oak.spi.commit.BackgroundObserverMBean; import org.apache.jackrabbit.oak.spi.commit.CommitInfo; import org.apache.jackrabbit.oak.spi.commit.Observer; import org.apache.jackrabbit.oak.spi.state.NodeState; @@ -81,6 +86,14 @@ class ChangeProcessor implements Observer { */ public static final int MAX_DELAY = 10000; + private static final AtomicInteger COUNTER = new AtomicInteger(); + + /** + * JMX ObjectName property storing the listenerId which allows + * to correlate various mbeans + */ + static final String LISTENER_ID = "listenerId"; + private final ContentSession contentSession; private final NamePathMapper namePathMapper; private final ListenerTracker tracker; @@ -134,10 +147,18 @@ class ChangeProcessor implements Observer { final WhiteboardExecutor executor = new WhiteboardExecutor(); executor.start(whiteboard); final BackgroundObserver observer = createObserver(executor); + Map attrs = ImmutableMap.of(LISTENER_ID, String.valueOf(COUNTER.incrementAndGet())); + String name = tracker.toString(); registration = new CompositeRegistration( registerObserver(whiteboard, observer), registerMBean(whiteboard, EventListenerMBean.class, - tracker.getListenerMBean(), "EventListener", tracker.toString()), + tracker.getListenerMBean(), "EventListener", name, attrs), + registerMBean(whiteboard, BackgroundObserverMBean.class, + observer.getMBean(), BackgroundObserverMBean.TYPE, name, attrs), + //TODO If FilterProvider gets changed later then MBean would need to be + // re-registered + registerMBean(whiteboard, FilterConfigMBean.class, + filterProvider.get().getConfigMBean(), FilterConfigMBean.TYPE, name, attrs), new Registration() { @Override public void unregister() { diff --git oak-jcr/src/main/java/org/apache/jackrabbit/oak/jcr/observation/ConsolidatedListenerMBeanImpl.java oak-jcr/src/main/java/org/apache/jackrabbit/oak/jcr/observation/ConsolidatedListenerMBeanImpl.java new file mode 100644 index 0000000..485a587 --- /dev/null +++ oak-jcr/src/main/java/org/apache/jackrabbit/oak/jcr/observation/ConsolidatedListenerMBeanImpl.java @@ -0,0 +1,422 @@ + + +/* + * 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.jackrabbit.oak.jcr.observation; + +import java.util.Arrays; +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.atomic.AtomicInteger; + +import javax.management.ObjectName; +import javax.management.openmbean.CompositeDataSupport; +import javax.management.openmbean.CompositeType; +import javax.management.openmbean.OpenDataException; +import javax.management.openmbean.OpenType; +import javax.management.openmbean.SimpleType; +import javax.management.openmbean.TabularData; +import javax.management.openmbean.TabularDataSupport; +import javax.management.openmbean.TabularType; + +import com.google.common.collect.Lists; +import com.google.common.collect.Maps; +import org.apache.felix.scr.annotations.Activate; +import org.apache.felix.scr.annotations.Component; +import org.apache.felix.scr.annotations.Deactivate; +import org.apache.felix.scr.annotations.Reference; +import org.apache.felix.scr.annotations.ReferenceCardinality; +import org.apache.felix.scr.annotations.ReferencePolicy; +import org.apache.felix.scr.annotations.References; +import org.apache.jackrabbit.api.jmx.EventListenerMBean; +import org.apache.jackrabbit.oak.jcr.observation.jmx.ConsolidatedListenerMBean; +import org.apache.jackrabbit.oak.osgi.OsgiWhiteboard; +import org.apache.jackrabbit.oak.plugins.observation.filter.FilterConfigMBean; +import org.apache.jackrabbit.oak.spi.commit.BackgroundObserverMBean; +import org.apache.jackrabbit.oak.spi.commit.Observer; +import org.apache.jackrabbit.oak.spi.whiteboard.Registration; +import org.apache.jackrabbit.oak.spi.whiteboard.Whiteboard; +import org.osgi.framework.BundleContext; + +import static com.google.common.base.Preconditions.checkNotNull; +import static org.apache.jackrabbit.oak.spi.whiteboard.WhiteboardUtils.registerMBean; + +@Component +@References({ + @Reference(name = "observer", + bind = "bindObserver", + unbind = "unbindObserver", + referenceInterface = Observer.class, + policy = ReferencePolicy.DYNAMIC, + cardinality = ReferenceCardinality.OPTIONAL_MULTIPLE), + @Reference(name = "listenerMBean", + bind = "bindListenerMBean", + unbind = "unbindListenerMBean", + referenceInterface = EventListenerMBean.class, + policy = ReferencePolicy.DYNAMIC, + cardinality = ReferenceCardinality.OPTIONAL_MULTIPLE), + @Reference(name = "backgroundObserverMBean", + bind = "bindBackgroundObserverMBean", + unbind = "unbindBackgroundObserverMBean", + referenceInterface = BackgroundObserverMBean.class, + policy = ReferencePolicy.DYNAMIC, + cardinality = ReferenceCardinality.OPTIONAL_MULTIPLE), + @Reference(name = "filterConfigMBean", + bind = "bindFilterConfigMBean", + unbind = "unbindFilterConfigMBean", + referenceInterface = FilterConfigMBean.class, + policy = ReferencePolicy.DYNAMIC, + cardinality = ReferenceCardinality.OPTIONAL_MULTIPLE) + +}) +public class ConsolidatedListenerMBeanImpl implements ConsolidatedListenerMBean { + private final AtomicInteger observerCount = new AtomicInteger(); + private final Map eventListeners = Maps.newConcurrentMap(); + private final Map bgObservers = Maps.newConcurrentMap(); + private final Map filterConfigs = Maps.newConcurrentMap(); + + private Registration mbeanReg; + + @Override + public TabularData getListenerStats() { + TabularDataSupport tds; + try { + int id = 0; + TabularType tt = new TabularType(ListenerStatsData.class.getName(), + "Consolidated Listener Stats", ListenerStatsData.TYPE, new String[]{"index"}); + tds = new TabularDataSupport(tt); + for(ListenerMBeans beans : getListenerMBeans()){ + tds.put(new ListenerStatsData(++id, beans).toCompositeData()); + } + } catch (OpenDataException e) { + throw new IllegalStateException(e); + } + return tds; + } + + @Override + public TabularData getObserversStats() { + TabularDataSupport tds; + try { + int id = 0; + TabularType tt = new TabularType(ObserverStatsData.class.getName(), + "Consolidated Observer Stats", ObserverStatsData.TYPE, new String[]{"index"}); + tds = new TabularDataSupport(tt); + for(BackgroundObserverMBean o: collectNonJcrObservers()){ + tds.put(new ObserverStatsData(++id, o).toCompositeData()); + } + } catch (OpenDataException e) { + throw new IllegalStateException(e); + } + return tds; + } + + @Override + public int getObserversCount() { + return observerCount.get(); + } + + @Override + public int getListenersCount() { + return eventListeners.size(); + } + + private Collection collectNonJcrObservers() { + List observers = Lists.newArrayList(); + + for (Map.Entry o : bgObservers.entrySet()){ + String listenerId = getListenerId(o.getKey()); + if (listenerId == null){ + observers.add(o.getValue()); + } + } + + return observers; + } + + /** + * Performs the mapping between EventListenerMBean and the Observer + * based on the JMX ObjectName service property of the MBean + * + * @return map of EventListenerMBean and corresponding Observer + */ + private List getListenerMBeans() { + List mbeans = Lists.newArrayListWithCapacity(eventListeners.size()); + for (Map.Entry e : eventListeners.entrySet()){ + String listenerId = getListenerId(e.getKey()); + ListenerMBeans m = new ListenerMBeans(); + m.eventListenerMBean = e.getValue(); + for (Map.Entry ef : filterConfigs.entrySet()){ + if (Objects.equals(getListenerId(ef.getKey()), listenerId)){ + m.filterConfigMBean = ef.getValue(); + } + } + for (Map.Entry ef : bgObservers.entrySet()){ + if (Objects.equals(getListenerId(ef.getKey()), listenerId)){ + m.observerMBean = ef.getValue(); + } + } + mbeans.add(m); + } + return mbeans; + } + + //~---------------------------------------< OSGi > + + @Activate + private void activate(BundleContext context){ + Whiteboard wb = new OsgiWhiteboard(context); + mbeanReg = registerMBean(wb, + ConsolidatedListenerMBean.class, + this, + ConsolidatedListenerMBean.TYPE, + "Consolidated Event Listener statistics"); + } + + @Deactivate + private void deactivate(){ + if(mbeanReg != null){ + mbeanReg.unregister(); + } + eventListeners.clear(); + bgObservers.clear(); + filterConfigs.clear(); + } + + @SuppressWarnings("unused") + protected void bindObserver(Observer observer, Map config){ + observerCount.incrementAndGet(); + } + + @SuppressWarnings("unused") + protected synchronized void unbindObserver(Observer observer, Map config){ + observerCount.decrementAndGet(); + } + + @SuppressWarnings("unused") + protected void bindBackgroundObserverMBean(BackgroundObserverMBean mbean, Map config){ + bgObservers.put(getObjectName(config), mbean); + } + + @SuppressWarnings("unused") + protected void unbindBackgroundObserverMBean(BackgroundObserverMBean mbean, Map config){ + bgObservers.remove(getObjectName(config)); + } + + @SuppressWarnings("unused") + protected void bindListenerMBean(EventListenerMBean mbean, Map config){ + eventListeners.put(getObjectName(config), mbean); + } + + @SuppressWarnings("unused") + protected void unbindListenerMBean(EventListenerMBean mbean, Map config){ + eventListeners.remove(getObjectName(config)); + } + + @SuppressWarnings("unused") + protected void bindFilterConfigMBean(FilterConfigMBean mbean, Map config){ + filterConfigs.put(getObjectName(config), mbean); + } + + @SuppressWarnings("unused") + protected void unbindFilterConfigMBean(FilterConfigMBean mbean, Map config){ + filterConfigs.remove(getObjectName(config)); + } + + private static ObjectName getObjectName(Map config){ + return checkNotNull((ObjectName) config.get("jmx.objectname"), + "No 'jmx.objectname' property defined for MBean %s", config); + } + + private static String getListenerId(ObjectName name){ + return name.getKeyProperty(ChangeProcessor.LISTENER_ID); + } + + private static class ListenerMBeans { + EventListenerMBean eventListenerMBean; + BackgroundObserverMBean observerMBean; + FilterConfigMBean filterConfigMBean; + } + + //~------------------------------------------< JMX > + + private static class ListenerStatsData { + static final String[] FIELD_NAMES = new String[]{ + "index", + "className", + "isDeep", + "nodeTypeNames", + "deliveries", + "deliveries/hr", + "us/delivery", + "delivered", + "delivered/hr", + "us/delivered", + "ratioOfTimeSpentProcessingEvents", + "queueSize", + "localEventCount", + "externalEventCount", + "paths", + "clusterExternal", + "clusterLocal", + "maxQueueSize" + }; + + static final String[] FIELD_DESCRIPTIONS = FIELD_NAMES; + + @SuppressWarnings("rawtypes") + static final OpenType[] FIELD_TYPES = new OpenType[]{ + SimpleType.INTEGER, + SimpleType.STRING, + SimpleType.BOOLEAN, + SimpleType.STRING, + SimpleType.LONG, + SimpleType.LONG, + SimpleType.LONG, + SimpleType.LONG, + SimpleType.LONG, + SimpleType.LONG, + SimpleType.DOUBLE, + SimpleType.INTEGER, + SimpleType.INTEGER, + SimpleType.INTEGER, + SimpleType.STRING, + SimpleType.BOOLEAN, + SimpleType.BOOLEAN, + SimpleType.INTEGER, + }; + + static final CompositeType TYPE = createCompositeType(); + + static CompositeType createCompositeType() { + try { + return new CompositeType( + ListenerStatsData.class.getName(), + "Composite data type for Listener statistics", + ListenerStatsData.FIELD_NAMES, + ListenerStatsData.FIELD_DESCRIPTIONS, + ListenerStatsData.FIELD_TYPES); + } catch (OpenDataException e) { + throw new IllegalStateException(e); + } + } + + private final ListenerMBeans mbeans; + private final int index; + + public ListenerStatsData(int i, ListenerMBeans mbeans){ + this.index = i; + this.mbeans = mbeans; + } + + CompositeDataSupport toCompositeData() { + Object[] values = new Object[]{ + index, + mbeans.eventListenerMBean.getClassName(), + mbeans.eventListenerMBean.isDeep(), + Arrays.toString(mbeans.eventListenerMBean.getNodeTypeName()), + mbeans.eventListenerMBean.getEventDeliveries(), + mbeans.eventListenerMBean.getEventDeliveriesPerHour(), + mbeans.eventListenerMBean.getMicrosecondsPerEventDelivery(), + mbeans.eventListenerMBean.getEventsDelivered(), + mbeans.eventListenerMBean.getEventsDeliveredPerHour(), + mbeans.eventListenerMBean.getMicrosecondsPerEventDelivered(), + mbeans.eventListenerMBean.getRatioOfTimeSpentProcessingEvents(), + mbeans.observerMBean.getQueueSize(), + mbeans.observerMBean.getLocalEventCount(), + mbeans.observerMBean.getExternalEventCount(), + Arrays.toString(mbeans.filterConfigMBean.getSubTrees()), + mbeans.filterConfigMBean.isIncludeClusterExternal(), + mbeans.filterConfigMBean.isIncludeClusterLocal(), + mbeans.observerMBean.getMaxQueueSize(), + }; + try { + return new CompositeDataSupport(TYPE, FIELD_NAMES, values); + } catch (OpenDataException e) { + throw new IllegalStateException(e); + } + } + } + + private static class ObserverStatsData { + static final String[] FIELD_NAMES = new String[]{ + "index", + "className", + "maxQueueSize", + "queueSize", + "localEventCount", + "externalEventCount", + }; + + static final String[] FIELD_DESCRIPTIONS = FIELD_NAMES; + + @SuppressWarnings("rawtypes") + static final OpenType[] FIELD_TYPES = new OpenType[]{ + SimpleType.INTEGER, + SimpleType.STRING, + SimpleType.INTEGER, + SimpleType.INTEGER, + SimpleType.INTEGER, + SimpleType.INTEGER, + }; + + static final CompositeType TYPE = createCompositeType(); + + static CompositeType createCompositeType() { + try { + return new CompositeType( + ObserverStatsData.class.getName(), + "Composite data type for Observer statistics", + ObserverStatsData.FIELD_NAMES, + ObserverStatsData.FIELD_DESCRIPTIONS, + ObserverStatsData.FIELD_TYPES); + } catch (OpenDataException e) { + throw new IllegalStateException(e); + } + } + + private final int index; + private BackgroundObserverMBean mbean; + + public ObserverStatsData(int i, BackgroundObserverMBean observer){ + this.index = i; + this.mbean = observer; + } + + CompositeDataSupport toCompositeData() { + Object[] values = new Object[]{ + index, + mbean.getClassName(), + mbean.getMaxQueueSize(), + mbean.getQueueSize(), + mbean.getLocalEventCount(), + mbean.getExternalEventCount(), + + }; + try { + return new CompositeDataSupport(TYPE, FIELD_NAMES, values); + } catch (OpenDataException e) { + throw new IllegalStateException(e); + } + } + } +} diff --git oak-jcr/src/main/java/org/apache/jackrabbit/oak/jcr/observation/jmx/ConsolidatedListenerMBean.java oak-jcr/src/main/java/org/apache/jackrabbit/oak/jcr/observation/jmx/ConsolidatedListenerMBean.java new file mode 100644 index 0000000..92bff2f --- /dev/null +++ oak-jcr/src/main/java/org/apache/jackrabbit/oak/jcr/observation/jmx/ConsolidatedListenerMBean.java @@ -0,0 +1,34 @@ +/* + * 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.jackrabbit.oak.jcr.observation.jmx; + +import javax.management.openmbean.TabularData; + +public interface ConsolidatedListenerMBean { + String TYPE = "ConsolidatedListenerStats"; + + TabularData getListenerStats(); + + TabularData getObserversStats(); + + int getListenersCount(); + + int getObserversCount(); +} diff --git oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexProviderService.java oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexProviderService.java index 48d46a9..116125c 100644 --- oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexProviderService.java +++ oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexProviderService.java @@ -41,6 +41,7 @@ import org.apache.jackrabbit.oak.osgi.OsgiWhiteboard; import org.apache.jackrabbit.oak.plugins.index.aggregate.NodeAggregator; import org.apache.jackrabbit.oak.spi.commit.BackgroundObserver; import org.apache.jackrabbit.oak.plugins.index.lucene.score.ScorerProviderFactory; +import org.apache.jackrabbit.oak.spi.commit.BackgroundObserverMBean; import org.apache.jackrabbit.oak.spi.commit.Observer; import org.apache.jackrabbit.oak.spi.query.QueryIndexProvider; import org.apache.jackrabbit.oak.spi.whiteboard.Registration; @@ -218,6 +219,11 @@ public class LuceneIndexProviderService { if (enableAsyncIndexOpen) { backgroundObserver = new BackgroundObserver(indexProvider, executor, 5); observer = backgroundObserver; + oakRegs.add(registerMBean(whiteboard, + BackgroundObserverMBean.class, + backgroundObserver.getMBean(), + BackgroundObserverMBean.TYPE, + "LuceneIndexConfigObserver queue stats")); log.info("Registering the LuceneIndexProvider as a BackgroundObserver"); } regs.add(bundleContext.registerService(Observer.class.getName(), observer, null));