ISPN-2861 Integrate the CloudTM extended statistics in Infinispan

README-i18n.md

@@ -29,3 +29,7 @@ Infinispan logs fully support internationalization. The following is a list of m
     17001 - 18000    cdi integration
     18001 - 19000    hbase
     21001 - 22000    mongodb cachestore
+    22001 - 23000    jpa cache store
+    23001 - 24000    leveldb cache store
+    24001 - 25000    couchbase cache store
+    25001 - 26000    extended statistics

core/src/main/java/org/infinispan/commands/remote/ClusteredGetCommand.java

@@ -55,6 +55,8 @@ public class ClusteredGetCommand extends BaseRpcCommand implements FlagAffectedC
    private InternalEntryFactory entryFactory;
    private int topologyId;
    private Equivalence keyEquivalence;
+   //only used by extended statistics. this boolean is local.
+   private boolean isWrite;
 
    private ClusteredGetCommand() {
       super(null); // For command id uniqueness test
@@ -72,6 +74,7 @@ public ClusteredGetCommand(Object key, String cacheName, Set<Flag> flags,
       this.acquireRemoteLock = acquireRemoteLock;
       this.gtx = gtx;
       this.keyEquivalence = keyEquivalence;
+      this.isWrite = false;
       if (acquireRemoteLock && (gtx == null))
          throw new IllegalArgumentException("Cannot have null tx if we need to acquire locks");
    }
@@ -120,6 +123,10 @@ public InternalCacheValue perform(InvocationContext context) throws Throwable {
       }
    }
 
+   public GlobalTransaction getGlobalTransaction() {
+      return gtx;
+   }
+
    private void acquireLocksIfNeeded() throws Throwable {
       if (acquireRemoteLock) {
          LockControlCommand lockControlCommand = commandsFactory.buildLockControlCommand(key, flags, gtx);
@@ -135,7 +142,7 @@ public byte getCommandId() {
 
    @Override
    public Object[] getParameters() {
-      return new Object[]{key, flags, acquireRemoteLock, gtx};
+      return new Object[]{key, flags, acquireRemoteLock, acquireRemoteLock ? gtx : null};
    }
 
    @Override
@@ -178,6 +185,14 @@ public String toString() {
          .toString();
    }
 
+   public boolean isWrite() {
+      return isWrite;
+   }
+
+   public void setWrite(boolean write) {
+      isWrite = write;
+   }
+
    public Object getKey() {
       return key;
    }

core/src/main/java/org/infinispan/commands/remote/recovery/TxCompletionNotificationCommand.java

@@ -97,6 +97,10 @@ public Object perform(InvocationContext ctx) throws Throwable {
       return null;
    }
 
+   public GlobalTransaction getGlobalTransaction() {
+      return gtx;
+   }
+
    /**
     * This only happens during state transfer.
     */

core/src/main/java/org/infinispan/container/entries/RepeatableReadEntry.java

@@ -48,14 +48,14 @@ public void performLocalWriteSkewCheck(DataContainer container, boolean alreadyC
       // the implicit "versioning" we have in R_R creates a new wrapper "value" instance for every update.
       if (actualValue != null && actualValue != valueToCompare) {
          log.unableToCopyEntryForUpdate(getKey());
-         throw new WriteSkewException("Detected write skew.");
+         throw new WriteSkewException("Detected write skew.", key);
       }
 
       if (valueToCompare != null && ice == null && !isCreated()) {
          // We still have a write-skew here.  When this wrapper was created there was an entry in the data container
          // (hence isCreated() == false) but 'ice' is now null.
          log.unableToCopyEntryForUpdate(getKey());
-         throw new WriteSkewException("Detected write skew - concurrent removal of entry!");
+         throw new WriteSkewException("Detected write skew - concurrent removal of entry!", key);
       }
    }
 

core/src/main/java/org/infinispan/interceptors/ClusteringInterceptor.java

@@ -94,7 +94,8 @@ protected boolean ignorePreviousValueOnBackup(WriteCommand command, InvocationCo
     * rehash in progress, involving nodes that the key maps to.
     *
     * @param key key to look up
+    * @param isWrite {@code true} if this is triggered by a write operation
     * @return an internal cache entry, or null if it cannot be located
     */
-   protected abstract InternalCacheEntry retrieveFromRemoteSource(Object key, InvocationContext ctx, boolean acquireRemoteLock, FlagAffectedCommand command) throws Exception;
+   protected abstract InternalCacheEntry retrieveFromRemoteSource(Object key, InvocationContext ctx, boolean acquireRemoteLock, FlagAffectedCommand command, boolean isWrite) throws Exception;
 }

core/src/main/java/org/infinispan/interceptors/distribution/BaseDistributionInterceptor.java

@@ -56,9 +56,10 @@ public void injectDependencies(DistributionManager distributionManager, Clusteri
    }
 
    @Override
-   protected final InternalCacheEntry retrieveFromRemoteSource(Object key, InvocationContext ctx, boolean acquireRemoteLock, FlagAffectedCommand command) throws Exception {
-      GlobalTransaction gtx = acquireRemoteLock ? ((TxInvocationContext)ctx).getGlobalTransaction() : null;
+   protected final InternalCacheEntry retrieveFromRemoteSource(Object key, InvocationContext ctx, boolean acquireRemoteLock, FlagAffectedCommand command, boolean isWrite) throws Exception {
+      GlobalTransaction gtx = ctx.isInTxScope() ? ((TxInvocationContext)ctx).getGlobalTransaction() : null;
       ClusteredGetCommand get = cf.buildClusteredGetCommand(key, command.getFlags(), acquireRemoteLock, gtx);
+      get.setWrite(isWrite);
 
       List<Address> targets = new ArrayList<Address>(stateTransferManager.getCacheTopology().getReadConsistentHash().locateOwners(key));
       // if any of the recipients has left the cluster since the command was issued, just don't wait for its response

core/src/main/java/org/infinispan/interceptors/distribution/NonTxDistributionInterceptor.java

@@ -152,7 +152,7 @@ private Object remoteGetBeforeWrite(InvocationContext ctx, WriteCommand command,
          if (trace) log.tracef("Doing a remote get for key %s", key);
 
          // attempt a remote lookup
-         InternalCacheEntry ice = retrieveFromRemoteSource(key, ctx, false, command);
+         InternalCacheEntry ice = retrieveFromRemoteSource(key, ctx, false, command, true);
          if (ice != null) {
             if (!ctx.replaceValue(key, ice)) {
                entryFactory.wrapEntryForPut(ctx, key, ice, false, command, true);
@@ -181,7 +181,7 @@ private InternalCacheEntry localGetCacheEntry(InvocationContext ctx, Object key,
 
    private InternalCacheEntry remoteGetCacheEntry(InvocationContext ctx, Object key, GetKeyValueCommand command) throws Throwable {
       if (trace) log.tracef("Doing a remote get for key %s", key);
-      InternalCacheEntry ice = retrieveFromRemoteSource(key, ctx, false, command);
+      InternalCacheEntry ice = retrieveFromRemoteSource(key, ctx, false, command, false);
       command.setRemotelyFetchedValue(ice);
       if (ice != null)
          return ice;

core/src/main/java/org/infinispan/interceptors/distribution/TxDistributionInterceptor.java

@@ -367,7 +367,7 @@ private Object remoteGetAndStoreInL1(InvocationContext ctx, Object key, boolean
             acquireRemoteLock = isWrite && isPessimisticCache && !txContext.getAffectedKeys().contains(key);
          }
          // attempt a remote lookup
-         InternalCacheEntry ice = retrieveFromRemoteSource(key, ctx, acquireRemoteLock, command);
+         InternalCacheEntry ice = retrieveFromRemoteSource(key, ctx, acquireRemoteLock, command, isWrite);
 
          if (acquireRemoteLock) {
             ((TxInvocationContext) ctx).addAffectedKey(key);

core/src/main/java/org/infinispan/transaction/WriteSkewException.java

@@ -9,18 +9,29 @@
  * @since 5.1
  */
 public class WriteSkewException extends CacheException {
+
+   private final Object key;
+
    public WriteSkewException() {
+      this.key = null;
    }
 
-   public WriteSkewException(Throwable cause) {
+   public WriteSkewException(Throwable cause, Object key) {
       super(cause);
+      this.key = key;
    }
 
-   public WriteSkewException(String msg) {
+   public WriteSkewException(String msg, Object key) {
       super(msg);
+      this.key = key;
    }
 
-   public WriteSkewException(String msg, Throwable cause) {
+   public WriteSkewException(String msg, Throwable cause, Object key) {
       super(msg, cause);
+      this.key = key;
+   }
+
+   public final Object getKey() {
+      return key;
    }
 }

core/src/main/java/org/infinispan/transaction/WriteSkewHelper.java

@@ -46,7 +46,7 @@ public static EntryVersionsMap performWriteSkewCheckAndReturnNewVersions(Version
                   uv.put(k, newVersion);
                } else {
                   // Write skew check detected!
-                  throw new WriteSkewException("Write skew detected on key " + k + " for transaction " + context.getTransaction());
+                  throw new WriteSkewException("Write skew detected on key " + k + " for transaction " + context.getTransaction(), k);
                }
             }
          }
@@ -70,7 +70,7 @@ public static EntryVersionsMap performTotalOrderWriteSkewCheckAndReturnNewVersio
                   uv.put(k, null);
                } else {
                   // Write skew check detected!
-                  throw new WriteSkewException("Write skew detected on key " + k + " for transaction " + context.getTransaction());
+                  throw new WriteSkewException("Write skew detected on key " + k + " for transaction " + context.getTransaction(), k);
                }
             }
          }

extended-statistics/pom.xml

@@ -0,0 +1,37 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.infinispan</groupId>
+        <artifactId>infinispan-parent</artifactId>
+        <version>6.0.0-SNAPSHOT</version>
+        <relativePath>../parent/pom.xml</relativePath>
+    </parent>
+
+    <artifactId>infinispan-extended-statistics</artifactId>
+    <packaging>jar</packaging>
+    <name>Infinispan Extended Statistics</name>
+    <description>Infinispan Extended Statistics module</description>
+
+    <dependencies>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>infinispan-core</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>infinispan-core</artifactId>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>com.clearspring.analytics</groupId>
+            <artifactId>stream</artifactId>
+            <version>2.2.0</version>
+        </dependency>
+    </dependencies>
+</project>