[QTL] Implement LookupExtractorFactory of namespaced lookup

extensions-core/namespace-lookup/pom.xml

@@ -77,5 +77,10 @@
       <version>3.0.1</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.easymock</groupId>
+      <artifactId>easymock</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 </project>

extensions-core/namespace-lookup/src/main/java/io/druid/query/extraction/NamespaceLookupExtractorFactory.java

@@ -0,0 +1,220 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  Metamarkets 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 io.druid.query.extraction;
+
+import com.fasterxml.jackson.annotation.JacksonInject;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import com.google.common.base.Preconditions;
+import com.metamx.common.ISE;
+import com.metamx.common.StringUtils;
+import com.metamx.common.logger.Logger;
+import io.druid.query.extraction.namespace.ExtractionNamespace;
+import io.druid.query.lookup.LookupExtractor;
+import io.druid.query.lookup.LookupExtractorFactory;
+import io.druid.query.lookup.LookupIntrospectHandler;
+import io.druid.server.namespace.cache.NamespaceExtractionCacheManager;
+
+import javax.annotation.Nullable;
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import java.nio.ByteBuffer;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+@JsonTypeName("namespace")
+public class NamespaceLookupExtractorFactory implements LookupExtractorFactory
+{
+  private static final Logger LOG = new Logger(NamespaceLookupExtractorFactory.class);
+
+  private static long SCHEDULE_TIMEOUT = 60_000;
+
+  private final AtomicBoolean started = new AtomicBoolean(false);
+  private final ReadWriteLock startStopSync = new ReentrantReadWriteLock();
+  private final ExtractionNamespace extractionNamespace;
+  private final NamespaceExtractionCacheManager manager;
+  private final LookupIntrospectHandler lookupIntrospectHandler;
+
+  private final String extractorID;
+
+  @JsonCreator
+  public NamespaceLookupExtractorFactory(
+      @JsonProperty("extractionNamespace") ExtractionNamespace extractionNamespace,
+      @JacksonInject NamespaceExtractionCacheManager manager
+  )
+  {
+    this.extractionNamespace = Preconditions.checkNotNull(
+        extractionNamespace,
+        "extractionNamespace should be specified"
+    );
+    this.manager = manager;
+    this.extractorID = buildID();
+    this.lookupIntrospectHandler = new LookupIntrospectHandler() {
+      @GET
+      @Path("/keys")
+      @Produces(MediaType.APPLICATION_JSON)
+      public Response getKeys()
+      {
+        return Response.ok(getLatest().keySet().toString()).build();
+      }
+
+      @GET
+      @Path("/values")
+      @Produces(MediaType.APPLICATION_JSON)
+      public Response getValues()
+      {
+        return Response.ok(getLatest().values().toString()).build();
+      }
+
+      @GET
+      @Produces(MediaType.APPLICATION_JSON)
+      public Response getMap()
+      {
+        return Response.ok(getLatest()).build();
+      }
+
+      private Map<String, String> getLatest()
+      {
+        return ((MapLookupExtractor)get()).getMap();
+      }
+    };
+  }
+
+  @Override
+  public boolean start()
+  {
+    final Lock writeLock = startStopSync.writeLock();
+    writeLock.lock();
+    try {
+      if (!started.compareAndSet(false, true)) {
+        LOG.warn("Already started!");
+        return true;
+      }
+      if (!manager.scheduleAndWait(extractorID, extractionNamespace, SCHEDULE_TIMEOUT)) {
+        LOG.warn("Failed to schedule lookup [%s]", extractorID);
+        return false;
+      }
+      LOG.debug("NamespaceLookupExtractorFactory[%s] started", extractorID);
+      return true;
+    }
+    finally {
+      writeLock.unlock();
+    }
+  }
+
+  @Override
+  public boolean close()
+  {
+    final Lock writeLock = startStopSync.writeLock();
+    writeLock.lock();
+    try {
+      if (!started.compareAndSet(true, false)) {
+        LOG.warn("Not started!");
+        return true;
+      }
+      return manager.checkedDelete(extractorID);
+    }
+    finally {
+      writeLock.unlock();
+    }
+  }
+
+  @Override
+  public boolean replaces(@Nullable LookupExtractorFactory other)
+  {
+    if (other != null && other instanceof NamespaceLookupExtractorFactory) {
+      NamespaceLookupExtractorFactory that = (NamespaceLookupExtractorFactory) other;
+      return !extractionNamespace.equals(that.extractionNamespace);
+    }
+    return true;
+  }
+
+  @Override
+  public LookupIntrospectHandler getIntrospectHandler()
+  {
+    return lookupIntrospectHandler;
+  }
+
+  @JsonProperty
+  public ExtractionNamespace getExtractionNamespace()
+  {
+    return extractionNamespace;
+  }
+
+  private String buildID()
+  {
+    return UUID.randomUUID().toString();
+  }
+
+  // Grab the latest snapshot from the cache manager
+  @Override
+  public LookupExtractor get()
+  {
+    final Lock readLock = startStopSync.readLock();
+    readLock.lock();
+    try {
+      if (!started.get()) {
+        throw new ISE("Factory [%s] not started", extractorID);
+      }
+      String preVersion = null, postVersion = null;
+      Map<String, String> map = null;
+      // Make sure we absolutely know what version of map we grabbed (for caching purposes)
+      do {
+        preVersion = manager.getVersion(extractorID);
+        if (preVersion == null) {
+          throw new ISE("Namespace vanished for [%s]", extractorID);
+        }
+        map = manager.getCacheMap(extractorID);
+        postVersion = manager.getVersion(extractorID);
+        if (postVersion == null) {
+          // We lost some horrible race... make sure we clean up
+          manager.delete(extractorID);
+          throw new ISE("Lookup [%s] is deleting", extractorID);
+        }
+      } while (!preVersion.equals(postVersion));
+      final byte[] v = StringUtils.toUtf8(postVersion);
+      final byte[] id = StringUtils.toUtf8(extractorID);
+      return new MapLookupExtractor(map, false)
+      {
+        @Override
+        public byte[] getCacheKey()
+        {
+          return ByteBuffer
+              .allocate(id.length + 1 + v.length + 1)
+              .put(id)
+              .put((byte) 0xFF)
+              .put(v)
+              .put((byte) 0xFF)
+              .array();
+        }
+      };
+    }
+    finally {
+      readLock.unlock();
+    }
+  }
+}

extensions-core/namespace-lookup/src/main/java/io/druid/query/extraction/namespace/ExtractionNamespace.java

@@ -34,11 +34,5 @@
  */
 public interface ExtractionNamespace
 {
-  /**
-   * This is expected to return the namespace name. As an additional requirement, the implementation MUST supply a
-   * "namespace" field in the json representing the object which is equal to the return of this function
-   * @return The name of the namespace
-   */
-  String getNamespace();
   long getPollMs();
 }

extensions-core/namespace-lookup/src/main/java/io/druid/query/extraction/namespace/ExtractionNamespaceFunctionFactory.java → extensions-core/namespace-lookup/src/main/java/io/druid/query/extraction/namespace/ExtractionNamespaceCacheFactory.java

@@ -28,29 +28,8 @@
 /**
  *
  */
-public interface ExtractionNamespaceFunctionFactory<T extends ExtractionNamespace>
+public interface ExtractionNamespaceCacheFactory<T extends ExtractionNamespace>
 {
-
-  /**
-   * Create a function for the given namespace which will do the manipulation requested in the extractionNamespace.
-   * A simple implementation would simply use the cache supplied by the `NamespaceExtractionCacheManager`.
-   * More advanced implementations may need more than just what can be cached by `NamespaceExtractionCacheManager`.
-   *
-   * @param extractionNamespace The ExtractionNamespace for which a manipulating function is needed.
-   *
-   * @return A function which will perform an extraction in accordance with the desires of the ExtractionNamespace
-   */
-  Function<String, String> buildFn(T extractionNamespace, Map<String, String> cache);
-
-
-  /**
-   * @param extractionNamespace The ExtractionNamespace for which a manipulating reverse function is needed.
-   * @param cache view of the cache containing the function mapping.
-   *
-   * @return A function that will perform reverse lookup.
-   */
-  Function<String, List<String>> buildReverseFn(T extractionNamespace, final Map<String, String> cache);
-
   /**
    * This function is called once if `ExtractionNamespace.getUpdateMs() == 0`, or every update if
    * `ExtractionNamespace.getUpdateMs() > 0`
@@ -60,6 +39,7 @@ public interface ExtractionNamespaceFunctionFactory<T extends ExtractionNamespac
    * initialize resources.
    * If the result of the Callable is the same as what is passed in as lastVersion, then no swap takes place, and the swap is discarded.
    *
+   * @param id                  The ID of ExtractionNamespace
    * @param extractionNamespace The ExtractionNamespace for which to populate data.
    * @param lastVersion         The version which was last cached
    * @param swap                The temporary Map into which data may be placed and will be "swapped" with the proper
@@ -70,5 +50,5 @@ public interface ExtractionNamespaceFunctionFactory<T extends ExtractionNamespac
    * @return A callable that will be used to refresh resources of the namespace and return the version string used in
    * the populating
    */
-  Callable<String> getCachePopulator(T extractionNamespace, String lastVersion, Map<String, String> swap);
+  Callable<String> getCachePopulator(String id, T extractionNamespace, String lastVersion, Map<String, String> swap);
 }