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..93409cf 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 @@ -391,6 +391,16 @@ public final class FilterBuilder { return subTrees; } + @Override + public boolean includeClusterLocal() { + return includeClusterLocal; + } + + @Override + public boolean includeClusterExternal() { + return includeClusterExternal; + } + private boolean isLocal(String sessionId, CommitInfo info) { return info != null && Objects.equal(info.getSessionId(), sessionId); } 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..b48bf7f 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,18 @@ public interface FilterProvider { */ @Nonnull Iterable getSubTrees(); + + /** + * Whether to include cluster local changes. + * + * @return true if cluster local changes need to be included + */ + boolean includeClusterLocal(); + + /** + * Whether to include cluster external changes. + * + * @return true if cluster external changes need to be included + */ + boolean includeClusterExternal(); } 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..396003a 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 @@ -202,6 +202,44 @@ public class BackgroundObserver implements Observer, Closeable { stopped = true; } + public Observer getDelegate(){ + return observer; + } + + public int queueSize(){ + return queue.size(); + } + + public QueueStats getQueueStats(){ + int size = queue.size(); + int localEventCount = 0; + int externalEventCount = 0; + int collapsedEventCount = 0; + + for (ContentChange cc : queue){ + if (cc.info != null){ + localEventCount++; + } else { + externalEventCount++; + } + } + return new QueueStats(size, localEventCount, externalEventCount, collapsedEventCount); + } + + public static final class QueueStats { + public final int size; + public final int localEventCount; + public final int externalEventCount; + public final int collapsedEventCount; + + public QueueStats(int size, int localEventCount, int externalEventCount, int collapsedEventCount) { + this.size = size; + this.localEventCount = localEventCount; + this.externalEventCount = externalEventCount; + this.collapsedEventCount = collapsedEventCount; + } + } + //----------------------------------------------------------< Observer >-- /** 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..56fd757 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 @@ -18,15 +18,16 @@ */ package org.apache.jackrabbit.oak.jcr.observation; +import static com.google.common.base.Preconditions.checkNotNull; import static com.google.common.base.Preconditions.checkState; import static org.apache.jackrabbit.api.stats.RepositoryStatistics.Type.OBSERVATION_EVENT_COUNTER; import static org.apache.jackrabbit.api.stats.RepositoryStatistics.Type.OBSERVATION_EVENT_DURATION; import static org.apache.jackrabbit.oak.plugins.observation.filter.VisibleFilter.VISIBLE_FILTER; import static org.apache.jackrabbit.oak.spi.whiteboard.WhiteboardUtils.registerMBean; -import static org.apache.jackrabbit.oak.spi.whiteboard.WhiteboardUtils.registerObserver; import static org.apache.jackrabbit.oak.spi.whiteboard.WhiteboardUtils.scheduleWithFixedDelay; 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 +37,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; @@ -81,6 +83,13 @@ class ChangeProcessor implements Observer { */ public static final int MAX_DELAY = 10000; + private static final AtomicInteger COUNTER = new AtomicInteger(); + + /** + * Service property name storing the Listener details + */ + static final String LISTENER_DETAILS = "listenerDetails"; + private final ContentSession contentSession; private final NamePathMapper namePathMapper; private final ListenerTracker tracker; @@ -134,10 +143,11 @@ class ChangeProcessor implements Observer { final WhiteboardExecutor executor = new WhiteboardExecutor(); executor.start(whiteboard); final BackgroundObserver observer = createObserver(executor); + String listenerDetails = tracker.toString() + ", listenerId " + COUNTER.incrementAndGet(); registration = new CompositeRegistration( - registerObserver(whiteboard, observer), + registerObserver(whiteboard, observer, listenerDetails), registerMBean(whiteboard, EventListenerMBean.class, - tracker.getListenerMBean(), "EventListener", tracker.toString()), + tracker.getListenerMBean(), "EventListener", listenerDetails), new Registration() { @Override public void unregister() { @@ -159,6 +169,10 @@ class ChangeProcessor implements Observer { ); } + FilterProvider getFilterProvider(){ + return filterProvider.get(); + } + private BackgroundObserver createObserver(final WhiteboardExecutor executor) { return new BackgroundObserver(this, executor, queueLength) { private volatile long delay; @@ -389,4 +403,12 @@ class ChangeProcessor implements Observer { return !wasStopped; } } + + private static Registration registerObserver( + @Nonnull Whiteboard whiteboard, + @Nonnull Observer observer, + String listenerDetails) { + return checkNotNull(whiteboard) + .register(Observer.class, checkNotNull(observer), ImmutableMap.of(LISTENER_DETAILS, listenerDetails)); + } } 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..d6e22bb --- /dev/null +++ oak-jcr/src/main/java/org/apache/jackrabbit/oak/jcr/observation/ConsolidatedListenerMBeanImpl.java @@ -0,0 +1,493 @@ + +/* + * 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.IdentityHashMap; +import java.util.List; +import java.util.Map; + +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.Iterables; +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.FilterProvider; +import org.apache.jackrabbit.oak.spi.commit.BackgroundObserver; +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 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 = "listener", + bind = "bindListener", + unbind = "unbindListener", + referenceInterface = EventListenerMBean.class, + policy = ReferencePolicy.DYNAMIC, + cardinality = ReferenceCardinality.OPTIONAL_MULTIPLE) +}) +public class ConsolidatedListenerMBeanImpl implements ConsolidatedListenerMBean { + private final Map> observeros = Maps.newIdentityHashMap(); + private final Map> listeners = Maps.newIdentityHashMap(); + private Registration mbeanReg; + + @Override + public TabularData getListenerStats() { + TabularDataSupport tds; + Map listenerObserverMapping + = getListenerObserverMapping(); + try { + int id = 0; + TabularType tt = new TabularType(ListenerStatsData.class.getName(), + "Consolidated Listener Stats", ListenerStatsData.TYPE, new String[]{"index"}); + tds = new TabularDataSupport(tt); + for(Map.Entry e: listenerObserverMapping.entrySet()){ + tds.put(new ListenerStatsData(++id, e.getKey(), e.getValue()).toCompositeData()); + } + } catch (OpenDataException e) { + throw new IllegalStateException(e); + } + return tds; + } + + @Override + public TabularData getObserversStats() { + TabularDataSupport tds; + Collection nonJcrObservers = collectNonJcrObservers(); + try { + int id = 0; + TabularType tt = new TabularType(ObserverStatsData.class.getName(), + "Consolidated Observer Stats", ObserverStatsData.TYPE, new String[]{"index"}); + tds = new TabularDataSupport(tt); + for(Observer o: nonJcrObservers){ + tds.put(new ObserverStatsData(++id, o).toCompositeData()); + } + } catch (OpenDataException e) { + throw new IllegalStateException(e); + } + return tds; + } + + @Override + public int getObserversCount() { + synchronized (observeros) { + return observeros.size(); + } + } + + @Override + public int getListenersCount() { + synchronized (listeners) { + return listeners.size(); + } + } + + private Collection collectNonJcrObservers() { + Map> observersCopy; + synchronized (observeros) { + observersCopy = new IdentityHashMap>(observeros); + } + + List observers = Lists.newArrayList(); + for (Map.Entry> o : observersCopy.entrySet()){ + String listenerName = (String) o.getValue().get(ChangeProcessor.LISTENER_DETAILS); + if (listenerName == null){ + observers.add(o.getKey()); + } + } + + 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 Map getListenerObserverMapping() { + Map mapping = Maps.newIdentityHashMap(); + + Map> observersCopy; + synchronized (observeros) { + observersCopy = new IdentityHashMap>(observeros); + } + + Map> listenersCopy; + synchronized (listeners) { + listenersCopy = new IdentityHashMap>(listeners); + } + + for (Map.Entry> e : listenersCopy.entrySet()){ + //Listener details are part of JMX bean object name "name" property + //The value is also quoted. See WhiteboardUtil#registerMBean + ObjectName oname = (ObjectName) e.getValue().get("jmx.objectname"); + String name = oname.getKeyProperty("name"); + + for (Map.Entry> o : observersCopy.entrySet()){ + String listenerName = (String) o.getValue().get(ChangeProcessor.LISTENER_DETAILS); + if (listenerName != null){ + listenerName = ObjectName.quote(listenerName); + if (name.equals(listenerName)){ + mapping.put(e.getKey(), o.getKey()); + break; + } + } + } + } + + return mapping; + } + + //~---------------------------------------< 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(); + } + observeros.clear(); + listeners.clear(); + } + + @SuppressWarnings("unused") + protected void bindObserver(Observer observer, Map config){ + synchronized (observeros){ + observeros.put(observer, config); + } + } + + @SuppressWarnings("unused") + protected synchronized void unbindObserver(Observer observer, Map config){ + synchronized (observeros){ + observeros.remove(observer); + } + } + + @SuppressWarnings("unused") + protected void bindListener(EventListenerMBean mbean, Map config){ + synchronized (listeners){ + listeners.put(mbean, config); + } + } + + @SuppressWarnings("unused") + protected void unbindListener(EventListenerMBean mbean, Map config){ + synchronized (listeners){ + listeners.remove(mbean); + } + } + + //~------------------------------------------< 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", + "collapsedEventCount", + "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.INTEGER, + SimpleType.STRING, + SimpleType.STRING, + SimpleType.STRING, + 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 EventListenerMBean stats; + private final int index; + private BackgroundObserver.QueueStats queueStats; + private FilterProvider filterProvider; + private BackgroundObserver backgroundObserver; + + public ListenerStatsData(int i, EventListenerMBean stats, Observer observer){ + this.index = i; + this.stats = stats; + + if (observer instanceof BackgroundObserver){ + backgroundObserver = (BackgroundObserver) observer; + queueStats = backgroundObserver.getQueueStats(); + Observer delegate = ((BackgroundObserver) observer).getDelegate(); + if (delegate instanceof ChangeProcessor){ + filterProvider = ((ChangeProcessor) delegate).getFilterProvider(); + } + } + } + + CompositeDataSupport toCompositeData() { + Object[] values = new Object[]{ + index, + stats.getClassName(), + stats.isDeep(), + Arrays.toString(stats.getNodeTypeName()), + stats.getEventDeliveries(), + stats.getEventDeliveriesPerHour(), + stats.getMicrosecondsPerEventDelivery(), + stats.getEventsDelivered(), + stats.getEventsDeliveredPerHour(), + stats.getMicrosecondsPerEventDelivered(), + stats.getRatioOfTimeSpentProcessingEvents(), + getQueueSize(), + getLocalEventCount(), + getExternalEventCount(), + getCollapsedEventCount(), + getPaths(), + getClusterExternal(), + getClusterLocal(), + getMaxQueueSize(), + }; + try { + return new CompositeDataSupport(TYPE, FIELD_NAMES, values); + } catch (OpenDataException e) { + throw new IllegalStateException(e); + } + } + + private int getMaxQueueSize() { + return backgroundObserver != null ? backgroundObserver.getMaxQueueLength() : -1; + } + + private Object getClusterLocal() { + if (filterProvider != null){ + return Boolean.toString(filterProvider.includeClusterLocal()); + } + return ""; + } + + private Object getClusterExternal() { + if (filterProvider != null){ + return Boolean.toString(filterProvider.includeClusterExternal()); + } + return ""; + } + + private String getPaths() { + if (filterProvider != null){ + return Iterables.toString(filterProvider.getSubTrees()); + } + return ""; + } + + private int getCollapsedEventCount() { + return queueStats != null ? queueStats.collapsedEventCount : -1; + } + + private int getExternalEventCount() { + return queueStats != null ? queueStats.externalEventCount : -1; + } + + private int getLocalEventCount() { + return queueStats != null ? queueStats.localEventCount : -1; + } + + private int getQueueSize() { + return queueStats != null ? queueStats.size : -1; + } + } + + private static class ObserverStatsData { + static final String[] FIELD_NAMES = new String[]{ + "index", + "className", + "maxQueueSize", + "queueSize", + "localEventCount", + "externalEventCount", + "collapsedEventCount", + }; + + 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, + 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 BackgroundObserver.QueueStats queueStats; + private Observer observer; + private BackgroundObserver backgroundObserver; + + public ObserverStatsData(int i, Observer observer){ + this.index = i; + this.observer = observer; + + if (observer instanceof BackgroundObserver){ + backgroundObserver = (BackgroundObserver) observer; + queueStats = backgroundObserver.getQueueStats(); + } + } + + CompositeDataSupport toCompositeData() { + Object[] values = new Object[]{ + index, + getClassName(), + getMaxQueueSize(), + getQueueSize(), + getLocalEventCount(), + getExternalEventCount(), + getCollapsedEventCount(), + + }; + try { + return new CompositeDataSupport(TYPE, FIELD_NAMES, values); + } catch (OpenDataException e) { + throw new IllegalStateException(e); + } + } + + private Object getClassName() { + if (backgroundObserver != null){ + return backgroundObserver.getDelegate().getClass().getName(); + } + return observer.getClass().getName(); + } + + private int getMaxQueueSize() { + return backgroundObserver != null ? backgroundObserver.getMaxQueueLength() : -1; + } + + private int getCollapsedEventCount() { + return queueStats != null ? queueStats.collapsedEventCount : -1; + } + + private int getExternalEventCount() { + return queueStats != null ? queueStats.externalEventCount : -1; + } + + private int getLocalEventCount() { + return queueStats != null ? queueStats.localEventCount : -1; + } + + private int getQueueSize() { + return queueStats != null ? queueStats.size : -1; + } + } +} 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(); +}