Add rest endpoints to dynamically add a catalog

presto-main/src/main/java/com/facebook/presto/dynamicCatalog/CatalogVo.java

@@ -0,0 +1,79 @@
+/*
+ * Licensed 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 com.facebook.presto.dynamicCatalog;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Map;
+
+public class CatalogVo
+{
+    @JsonProperty("catalogName")
+    public String catalogName;
+    @JsonProperty("connectorName")
+    public String connectorName;
+    @JsonProperty("properties")
+    public Map<String, String> properties;
+
+    public CatalogVo(String catalogName, String connectorName, Map<String, String> properties)
+    {
+        this.catalogName = catalogName;
+        this.connectorName = connectorName;
+        this.properties = properties;
+    }
+
+    public CatalogVo()
+    {
+    }
+
+    public String getCatalogName()
+    {
+        return catalogName;
+    }
+
+    public void setCatalogName(String catalogName)
+    {
+        this.catalogName = catalogName;
+    }
+
+    public String getConnectorName()
+    {
+        return connectorName;
+    }
+
+    public void setConnectorName(String connectorName)
+    {
+        this.connectorName = connectorName;
+    }
+
+    public Map<String, String> getProperties()
+    {
+        return properties;
+    }
+
+    public void setProperties(Map<String, String> properties)
+    {
+        this.properties = properties;
+    }
+
+    @Override
+    public String toString()
+    {
+        return "CatalogVo{" +
+                "catalogName='" + catalogName + '\'' +
+                ", connectorName='" + connectorName + '\'' +
+                ", properties=" + properties +
+                '}';
+    }
+}

presto-main/src/main/java/com/facebook/presto/dynamicCatalog/DynamicCatalogController.java

@@ -0,0 +1,179 @@
+/*
+ * Licensed 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 com.facebook.presto.dynamicCatalog;
+
+import com.facebook.presto.connector.ConnectorId;
+import com.facebook.presto.connector.ConnectorManager;
+import com.facebook.presto.metadata.CatalogManager;
+import com.facebook.presto.metadata.InternalNodeManager;
+import com.facebook.presto.metadata.StaticCatalogStoreConfig;
+import com.google.common.base.Joiner;
+import com.google.common.base.Splitter;
+import io.airlift.discovery.client.Announcer;
+import io.airlift.discovery.client.ServiceAnnouncement;
+import io.airlift.log.Logger;
+
+import javax.inject.Inject;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.DELETE;
+import javax.ws.rs.POST;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.QueryParam;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.OutputStream;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
+import static com.google.common.base.Strings.nullToEmpty;
+import static io.airlift.discovery.client.ServiceAnnouncement.serviceAnnouncement;
+
+@Path("/v1/catalog")
+public class DynamicCatalogController
+{
+    private final CatalogManager catalogManager;
+    private final ConnectorManager connectorManager;
+    private final Announcer announcer;
+    private final InternalNodeManager internalNodeManager;
+    private final File catalogConfigurationDir;
+    private static final Logger log = Logger.get(DynamicCatalogController.class);
+    private final ResponseParser responseParser;
+
+    private DynamicCatalogController(CatalogManager catalogManager, ConnectorManager connectorManager, Announcer announcer, InternalNodeManager internalNodeManager, File catalogConfigurationDir, ResponseParser responseParser)
+    {
+        this.catalogManager = catalogManager;
+        this.connectorManager = connectorManager;
+        this.announcer = announcer;
+        this.internalNodeManager = internalNodeManager;
+        this.catalogConfigurationDir = catalogConfigurationDir;
+        this.responseParser = responseParser;
+    }
+
+    @Inject
+    public DynamicCatalogController(CatalogManager catalogManager, ConnectorManager connectorManager, Announcer announcer, InternalNodeManager internalNodeManager, StaticCatalogStoreConfig config, ResponseParser responseParser)
+    {
+        this(catalogManager, connectorManager, announcer, internalNodeManager, config.getCatalogConfigurationDir(), responseParser);
+    }
+
+    @Path("add")
+    @POST
+    @Consumes(MediaType.APPLICATION_JSON)
+    @Produces(MediaType.APPLICATION_JSON)
+    public Response addCatalog(CatalogVo catalogVo)
+    {
+        try {
+            log.info("addCatalog : input values are " + catalogVo);
+            ConnectorId connectorId = connectorManager.createConnection(catalogVo.getCatalogName(), catalogVo.getConnectorName(), catalogVo.getProperties());
+            updateConnectorIdAnnouncement(announcer, connectorId, internalNodeManager);
+            log.info("addCatalog() : catalogConfigurationDir: " + catalogConfigurationDir.getAbsolutePath());
+            writeToFile(catalogVo);
+            log.info("addCatalog() : Successfully added catalog " + catalogVo.getCatalogName());
+            return successResponse(responseParser.build("Successfully added catalog: " + catalogVo.getCatalogName(), 200));
+        }
+        catch (Exception ex) {
+            log.error("addCatalog() : Error adding catalog " + ex.getMessage());
+            return failedResponse(responseParser.build("Error adding Catalog: " + ex.getMessage(), 500));
+        }
+    }
+
+    @Path("delete")
+    @DELETE
+    @Consumes(MediaType.APPLICATION_JSON)
+    @Produces(MediaType.APPLICATION_JSON)
+    public Response deleteCatalog(@QueryParam("catalogName") String catalogName)
+    {
+        log.info("deleteCatalog(): Deleting catalog: " + catalogName);
+        String responseMessage = "";
+        if (catalogManager.getCatalog(catalogName).isPresent()) {
+            log.info("deleteCatalog() : Catalog exists so deleting catalog " + catalogName);
+            connectorManager.dropConnection(catalogName);
+            responseMessage = (deletePropertyFile(catalogName) == true) ? "Successfully deleted" : "Error deleting catalog";
+        }
+        else {
+            log.info("deleteCatalog() : Catalog doesn't exists, Can't be deleted " + catalogName);
+            return failedResponse(responseParser.build("Catalog doesn't exists: " + catalogName, 500));
+        }
+        log.info("deleteCatalog() : successfully deleted catalog " + catalogName);
+        return successResponse(responseParser.build(responseMessage + " : " + catalogName, 200));
+    }
+
+    private boolean deletePropertyFile(String catalogName)
+    {
+        return new File(getPropertyFilePath(catalogName)).delete();
+    }
+
+    private void writeToFile(CatalogVo catalogVo)
+            throws Exception
+    {
+        final Properties properties = new Properties();
+        properties.put("connector.name", catalogVo.getConnectorName());
+        properties.putAll(catalogVo.getProperties());
+        String filePath = getPropertyFilePath(catalogVo.getCatalogName());
+        log.info("filepath: " + filePath);
+        File propertiesFile = new File(filePath);
+        try (OutputStream out = new FileOutputStream(propertiesFile)) {
+            properties.store(out, "adding catalog using endpoint");
+        }
+        catch (Exception ex) {
+            log.error("error while writing to a file :" + ex.getMessage());
+            throw new Exception("Error writing to file " + ex.getMessage());
+        }
+    }
+
+    private String getPropertyFilePath(String catalogName)
+    {
+        return catalogConfigurationDir.getPath() + File.separator + catalogName + ".properties";
+    }
+
+    private static ServiceAnnouncement getPrestoAnnouncement(Set<ServiceAnnouncement> announcements)
+    {
+        for (ServiceAnnouncement announcement : announcements) {
+            if (announcement.getType().equals("presto")) {
+                return announcement;
+            }
+        }
+        throw new IllegalArgumentException("Presto announcement not found: " + announcements);
+    }
+
+    private static void updateConnectorIdAnnouncement(Announcer announcer, ConnectorId connectorId, InternalNodeManager nodeManager)
+    {
+        ServiceAnnouncement announcement = getPrestoAnnouncement(announcer.getServiceAnnouncements());
+        Map<String, String> properties = new LinkedHashMap<>(announcement.getProperties());
+        String property = nullToEmpty(properties.get("connectorIds"));
+        Set<String> connectorIds = new LinkedHashSet<>(Splitter.on(',').trimResults().omitEmptyStrings().splitToList(property));
+        connectorIds.add(connectorId.toString());
+        properties.put("connectorIds", Joiner.on(',').join(connectorIds));
+        announcer.removeServiceAnnouncement(announcement.getId());
+        announcer.addServiceAnnouncement(serviceAnnouncement(announcement.getType()).addProperties(properties).build());
+        announcer.forceAnnounce();
+        nodeManager.refreshNodes();
+    }
+
+    private Response successResponse(ResponseParser responseParser)
+    {
+        return Response.status(Response.Status.OK).entity(responseParser).type(MediaType.APPLICATION_JSON).build();
+    }
+
+    private Response failedResponse(ResponseParser responseParser)
+    {
+        return Response.status(Response.Status.INTERNAL_SERVER_ERROR).entity(responseParser).type(MediaType.APPLICATION_JSON).build();
+    }
+}

presto-main/src/main/java/com/facebook/presto/dynamicCatalog/ResponseParser.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed 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 com.facebook.presto.dynamicCatalog;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public class ResponseParser
+{
+    @JsonProperty("message")
+    public String message;
+    @JsonProperty("statusCode")
+    public int statusCode;
+
+    public ResponseParser build(String message, int statusCode)
+    {
+        this.message = message;
+        this.statusCode = statusCode;
+        return this;
+    }
+
+    public ResponseParser()
+    {
+    }
+}

presto-main/src/main/java/com/facebook/presto/server/CoordinatorModule.java

@@ -21,6 +21,8 @@
 import com.facebook.presto.cost.CostComparator;
 import com.facebook.presto.cost.StatsCalculatorModule;
 import com.facebook.presto.cost.TaskCountEstimator;
+import com.facebook.presto.dynamicCatalog.DynamicCatalogController;
+import com.facebook.presto.dynamicCatalog.ResponseParser;
 import com.facebook.presto.event.QueryMonitor;
 import com.facebook.presto.event.QueryMonitorConfig;
 import com.facebook.presto.execution.AddColumnTask;
@@ -190,6 +192,9 @@ protected void setup(Binder binder)
         newExporter(binder).export(StatementResource.class).withGeneratedName();
         binder.bind(StatementHttpExecutionMBean.class).in(Scopes.SINGLETON);
         newExporter(binder).export(StatementHttpExecutionMBean.class).withGeneratedName();
+        jaxrsBinder(binder).bind(DynamicCatalogController.class);
+        newExporter(binder).export(DynamicCatalogController.class).withGeneratedName();
+        binder.bind(ResponseParser.class).in(Scopes.SINGLETON);
 
         // resource for serving static content
         jaxrsBinder(binder).bind(WebUiResource.class);