Add ability to auto-register GeoMesa File System DataStores

geomesa-gs-catalog-listener/pom.xml

@@ -16,6 +16,11 @@
             <artifactId>geomesa-accumulo-datastore_2.11</artifactId>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>org.locationtech.geomesa</groupId>
+            <artifactId>geomesa-fs-datastore_2.11</artifactId>
+            <scope>provided</scope>
+        </dependency>
         <dependency>
             <groupId>org.locationtech.geomesa</groupId>
             <artifactId>geomesa-web-core_2.11</artifactId>
@@ -41,6 +46,10 @@
             <artifactId>accumulo-core</artifactId>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>org.geoserver</groupId>
+            <artifactId>gs-ows</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

geomesa-gs-catalog-listener/src/main/resources/applicationContext.xml

@@ -7,4 +7,8 @@
         <!-- geoserver catalog -->
         <property name="catalog" ref="catalog"/>
     </bean>
+
+    <bean id="registerFilter" class="org.geomesa.gs.catalog.AutoRegisterFilter">
+        <property name="catalog" ref="catalog"/>
+    </bean>
 </beans>

geomesa-gs-catalog-listener/src/main/scala/org/geomesa/gs/catalog/AutoRegisterFilter.scala

@@ -0,0 +1,62 @@
+/***********************************************************************
+ * Copyright (c) 2013-2017 Commonwealth Computer Research, Inc.
+ * All rights reserved. This program and the accompanying materials are
+ * made available under the terms of the GNU GENERAL PUBLIC LICENSE,
+ * Version 2 which accompanies this distribution and is available at
+ * https://opensource.org/licenses/GPL-2.0.
+ ***********************************************************************/
+
+package org.geomesa.gs.catalog
+
+import javax.servlet.http.HttpServletRequest
+import javax.servlet.{FilterChain, FilterConfig, ServletRequest, ServletResponse}
+
+import org.apache.hadoop.fs.Path
+import org.geoserver.catalog.Catalog
+import org.geoserver.filters.GeoServerFilter
+import org.geoserver.ows.Request
+
+class AutoRegisterFilter extends GeoServerFilter {
+  var catalog: Catalog = _
+  def setCatalog(catalog: Catalog): Unit = this.catalog = catalog
+  def getCatalog: Catalog = this.catalog
+
+  override def init(filterConfig: FilterConfig): Unit = { }
+
+  override def doFilter(servletRequest: ServletRequest, servletResponse: ServletResponse, filterChain: FilterChain): Unit = {
+    val request = new Request()
+    request.setHttpRequest(servletRequest.asInstanceOf[HttpServletRequest])
+
+    val path = servletRequest.asInstanceOf[HttpServletRequest].getServletPath.split("/")
+    if (path.length > 1) {
+      createWorkspace(path(1))
+    }
+    filterChain.doFilter(servletRequest, servletResponse)
+  }
+
+  def createWorkspace(ws: String): Unit = {
+    val wsi = catalog.getWorkspaceByName(ws)
+    if (wsi == null && shouldCreate(ws)) {
+      val newWorkspace = catalog.getFactory.createWorkspace()
+      newWorkspace.setName(ws)
+
+      val namespace = catalog.getFactory.createNamespace()
+      namespace.setPrefix(newWorkspace.getName)
+      namespace.setURI(s"http://geomesa.org/$ws")
+
+      catalog.add(namespace)
+      catalog.add(newWorkspace)
+    }
+  }
+
+  def shouldCreate(workspace: String): Boolean = {
+    val base = System.getProperty("GEOMESA_FSDS_BASE_DIRECTORY")
+    if(base == null || Seq("styles", "web", "index.html", "openlayers3").contains(workspace)) return false
+
+    val path = new Path(base, workspace)
+    val fs = path.getFileSystem(new org.apache.hadoop.conf.Configuration)
+    fs.exists(path)
+  }
+
+  override def destroy(): Unit = { }
+}

geomesa-gs-catalog-listener/src/main/scala/org/geomesa/gs/catalog/GeoMesaCatalogListener.scala

@@ -9,11 +9,17 @@
 package org.geomesa.gs.catalog
 
 import com.typesafe.scalalogging.LazyLogging
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.{FileContext, FileStatus, Path, RemoteIterator}
 import org.geoserver.catalog.event._
-import org.geoserver.catalog.{Catalog, DataStoreInfo, FeatureTypeInfo, WorkspaceInfo}
+import org.geoserver.catalog._
 import org.geoserver.security.decorators.DecoratingDataStore
 import org.geotools.data.DataStore
-import org.locationtech.geomesa.accumulo.data.AccumuloDataStore
+import org.geotools.geometry.jts.ReferencedEnvelope
+import org.locationtech.geomesa.fs.FileSystemDataStoreFactory
+import org.locationtech.geomesa.fs.FileSystemDataStoreFactory.FileSystemDataStoreParams
+import org.locationtech.geomesa.fs.storage.common.utils.StorageUtils
+import org.locationtech.geomesa.index.geotools.GeoMesaDataStore
 import org.locationtech.geomesa.web.core.GeoMesaServletCatalog
 import org.locationtech.geomesa.web.core.GeoMesaServletCatalog.GeoMesaLayerInfo
 import org.springframework.beans.factory.InitializingBean
@@ -38,19 +44,18 @@ class GeoMesaCatalogListener extends CatalogListener with InitializingBean with
   def crawlCatalog(): Unit = catalog.getFeatureTypes.foreach(addFeatureTypeInfo)
 
   def addFeatureTypeInfo(featureTypeInfo: FeatureTypeInfo): Unit = {
-    // Handle ADS case
     getStore(featureTypeInfo) match {
-      case ads: AccumuloDataStore =>
+      case gmds: GeoMesaDataStore[_, _, _] =>
 
         val workspace = featureTypeInfo.getStore.getWorkspace.getName
         val layerName = featureTypeInfo.getName
 
         val nativeName = featureTypeInfo.getNativeName
 
-        val sft = ads.getSchema(nativeName)
+        val sft = gmds.getSchema(nativeName)
 
         logger.debug(s"Registering info for layer $layerName with SFT ${sft.getTypeName} with GeoMesa Stats REST API.")
-        GeoMesaServletCatalog.putGeoMesaLayerInfo(workspace, layerName, GeoMesaLayerInfo(ads, sft))
+        GeoMesaServletCatalog.putGeoMesaLayerInfo(workspace, layerName, GeoMesaLayerInfo(gmds, sft))
       case s => logger.debug(s"Encountered non-GeoMesa store: ${Option(s).map(_.getClass.getName).orNull}")
     }
   }
@@ -64,9 +69,8 @@ class GeoMesaCatalogListener extends CatalogListener with InitializingBean with
   }
 
   def removeFeatureTypeInfo(featureTypeInfo: FeatureTypeInfo): Unit = {
-    // Handle ADS case
     getStore(featureTypeInfo) match {
-      case ads: AccumuloDataStore =>
+      case gmds: GeoMesaDataStore[_, _, _] =>
 
         val workspace = featureTypeInfo.getStore.getWorkspace.getName
         val layerName = featureTypeInfo.getName
@@ -79,13 +83,34 @@ class GeoMesaCatalogListener extends CatalogListener with InitializingBean with
   }
 
   override def handleAddEvent(event: CatalogAddEvent): Unit = {
-    logger.debug(s"GeoMesa Catalog Listener received add event: $event")
+    logger.debug(s"GeoMesa Catalog Listener received add event: $event of type ${event.getSource.getClass}")
     event.getSource match {
       case fti: FeatureTypeInfo => addFeatureTypeInfo(fti)
-      case _ => // not a new layer; no action necessary.
+      case dti: DataStoreInfo => registerLayers(dti)
+      case wsi: WorkspaceInfo => registerDataStore(wsi)
+      case _ => event.getSource.getClass  // not a new layer; no action necessary.
     }
   }
 
+  def registerDataStore(wsi: WorkspaceInfo): Unit = {
+    val base = System.getProperty("GEOMESA_FSDS_BASE_DIRECTORY")
+    val directory = wsi.getName
+
+    val dsi = catalog.getFactory.createDataStore()
+    dsi.setWorkspace(wsi)
+    dsi.setEnabled(true)
+    dsi.setName(directory)
+
+    val factory = new FileSystemDataStoreFactory
+    dsi.setType(factory.getDisplayName())
+    val connParams = dsi.getConnectionParameters
+    val path = s"$base$directory"
+    connParams.put(FileSystemDataStoreParams.PathParam.getName, path)
+    connParams.put("namespace", catalog.getNamespaceByPrefix(wsi.getName).getURI)
+
+    catalog.save(dsi)
+  }
+
   override def handleRemoveEvent(event: CatalogRemoveEvent): Unit = {
     logger.debug(s"GeoMesa Catalog Listener received remove event: $event")
     event.getSource match {
@@ -109,6 +134,36 @@ class GeoMesaCatalogListener extends CatalogListener with InitializingBean with
     }
   }
 
+  // First pass of auto-registering layers.
+  def registerLayers(dti: DataStoreInfo): Unit = {
+    val ds = dti.getDataStore(null)
+    val wsi = dti.getWorkspace
+
+    ds.getNames.foreach { name =>
+      val fs = ds.getFeatureSource(name)
+
+      val builder: CatalogBuilder = new CatalogBuilder(catalog)
+      builder.setStore(dti)
+      val fti = builder.buildFeatureType(name)
+      val li = builder.buildLayer(fti)
+      val bbox = {
+        val tmp = builder.getNativeBounds(fti)
+        if (tmp.equals(ReferencedEnvelope.EVERYTHING)) {
+          builder.getBoundsFromCRS(fti)
+        } else {
+          tmp
+        }
+      }
+
+      fti.setNativeBoundingBox(bbox)
+      fti.setLatLonBoundingBox(bbox)
+      val namespace = catalog.getNamespaceByPrefix(wsi.getName)
+      fti.setNamespace(namespace)
+      catalog.add(fti)
+      catalog.add(li)
+    }
+  }
+
   def handleDataStoreUpdate(event: CatalogModifyEvent): Unit = {
     event.getSource match {
       case dsi: DataStoreInfo if event.getPropertyNames.contains("workspace") =>

pom.xml

@@ -445,6 +445,12 @@
                 <version>${slf4j.version}</version>
                 <scope>provided</scope>
             </dependency>
+            <dependency>
+                <groupId>org.specs2</groupId>
+                <artifactId>specs2_2.11</artifactId>
+                <version>${specs2.version}</version>
+                <scope>test</scope>
+            </dependency>
         </dependencies>
     </dependencyManagement>