Implemented the Rendler C++ Framework using V1 API.

cpp/Makefile

@@ -7,14 +7,22 @@ CXXCOMPILE = $(CXX) $(INCLUDES) $(CXXFLAGS) -c -o $@
 CXXLINK = $(CXX) $(INCLUDES) $(CXXFLAGS) -o $@
 
 default: all
-all: rendler crawl_executor render_executor
+all: rendler crawl_executor render_executor v1
 
 HEADERS = rendler_helper.hpp
-
+HEADERS_V1 = $(HEADERS) rendler_v1_executor.hpp
 
 crawl_executor: crawl_executor.cpp $(HEADERS)
 	$(CXXLINK) $<  $(LDFLAGS) -lboost_regex -lcurl
 
+rendler_v1_executor.o: rendler_v1_executor.cpp $(HEADERS_V1)
+	$(CXXLINK) -c $<
+
+%_v1_executor: %_v1_executor.cpp rendler_v1_executor.o $(HEADERS_V1)
+	$(CXXLINK) $< rendler_v1_executor.o $(LDFLAGS) -lboost_regex -lcurl
+
+v1: rendler_v1 crawl_v1_executor render_v1_executor
+
 %: %.cpp $(HEADERS)
 	$(CXXLINK) $< $(LDFLAGS)
 

cpp/crawl_v1_executor.cpp

@@ -0,0 +1,165 @@
+/**
+ * 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.
+ */
+
+#include <iostream>
+#include <vector>
+
+#include <boost/regex.hpp>
+
+#include <curl/curl.h>
+
+#include <stout/os.hpp>
+
+#include "rendler_helper.hpp"
+#include "rendler_v1_executor.hpp"
+
+using std::cout;
+using std::endl;
+using std::vector;
+
+using mesos::vectorToString;
+
+
+static int writer(char *data, size_t size, size_t nmemb, string *writerData)
+{
+  assert(writerData != NULL);
+  writerData->append(data, size*nmemb);
+  return size * nmemb;
+}
+
+
+class CrawlV1Executor : public RendlerV1Executor
+{
+public:
+  CrawlV1Executor(const FrameworkID& _frameworkId,
+                  const ExecutorID& _executorId)
+    : RendlerV1Executor("CrawlV1Executor", _frameworkId, _executorId) {}
+
+  virtual ~CrawlV1Executor() {}
+
+protected:
+  void runTask(const TaskInfo& task) override
+  {
+    string url = task.data();
+    cout << "Running crawl task " << task.task_id().value()
+         << " Fetch: " << url;
+
+    string buffer;
+    vector<string> result;
+    result.push_back(task.task_id().value());
+    result.push_back(url);
+
+    CURL *conn;
+    conn = curl_easy_init();
+    assert(conn != NULL);
+    assert(curl_easy_setopt(conn, CURLOPT_URL, url.c_str()) == CURLE_OK);
+    assert(curl_easy_setopt(conn, CURLOPT_FOLLOWLOCATION, 1L) == CURLE_OK);
+    assert(curl_easy_setopt(conn, CURLOPT_WRITEFUNCTION, writer) == CURLE_OK);
+    assert(curl_easy_setopt(conn, CURLOPT_WRITEDATA, &buffer) == CURLE_OK);
+
+    if (curl_easy_perform(conn) != CURLE_OK) {
+      return;
+    }
+
+    char *tmp;
+    assert(curl_easy_getinfo(conn, CURLINFO_EFFECTIVE_URL, &tmp) == CURLE_OK);
+    string redirectUrl = url;
+    if (tmp != NULL) {
+      redirectUrl = tmp;
+    }
+    curl_easy_cleanup(conn);
+
+    size_t scheme = redirectUrl.find_first_of("://");
+    size_t sp = redirectUrl.find_first_of('/',
+        scheme == string::npos ? 0 : scheme + 3); // skip the http:// part.
+    size_t lsp = redirectUrl.find_last_of('/'); // skip the http:// part.
+    string baseUrl = redirectUrl.substr(0, sp); // No trailing slash.
+    string dirUrl = redirectUrl.substr(0, lsp); // No trailing slash.
+
+    cout << "redirectUrl " << redirectUrl << " baseUrl: " << baseUrl << endl;
+    cout << "dirUrl " << dirUrl  << endl;
+
+    const boost::regex hrefRE("<a\\s+[^\\>]*?href\\s*=\\s*([\"'])(.*?)\\1");
+    const boost::regex urlRE("^([a-zA-Z]+://).*");
+
+    boost::smatch matchHref;
+    string::const_iterator f = buffer.begin();
+    string::const_iterator l = buffer.end();
+
+    while (f != buffer.end() &&
+           boost::regex_search(f, l, matchHref, hrefRE)) {
+      string link = matchHref[2];
+      f = matchHref[0].second;
+
+      boost::smatch matchService;
+      string::const_iterator lb = link.begin();
+      string::const_iterator le = link.end();
+
+      // Remove the anchor
+      if (link.find_first_of('#') != string::npos) {
+        link.erase(link.find_first_of('#'));
+      }
+
+      if (link.empty()) {
+        continue;
+      }
+
+      if (link[0] == '/') {
+        link = baseUrl + link;
+      } else if (!boost::regex_search(lb, le, matchService, urlRE)) {
+        // Relative URL
+        link = dirUrl + "/" + link;
+      }
+      result.push_back(link);
+    };
+
+    sendFrameworkMessage(vectorToString(result));
+    sendStatusUpdate(task, TASK_FINISHED);
+  }
+};
+
+
+int main(int argc, char** argv)
+{
+  FrameworkID frameworkId;
+  ExecutorID executorId;
+
+  Option<string> value;
+
+  value = os::getenv("MESOS_FRAMEWORK_ID");
+  if (value.isNone()) {
+    EXIT(EXIT_FAILURE)
+      << "Expecting 'MESOS_FRAMEWORK_ID' to be set in the environment";
+  }
+  frameworkId.set_value(value.get());
+
+  value = os::getenv("MESOS_EXECUTOR_ID");
+  if (value.isNone()) {
+    EXIT(EXIT_FAILURE)
+      << "Expecting 'MESOS_EXECUTOR_ID' to be set in the environment";
+  }
+  executorId.set_value(value.get());
+
+  process::Owned<CrawlV1Executor> crawler(
+      new CrawlV1Executor(frameworkId, executorId));
+
+  process::spawn(crawler.get());
+  process::wait(crawler.get());
+
+  return 0;
+}

cpp/render_v1_executor.cpp

@@ -0,0 +1,104 @@
+/**
+ * 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.
+ */
+
+#include <libgen.h>
+#include <iostream>
+#include <vector>
+
+#include <boost/regex.hpp>
+
+#include <curl/curl.h>
+
+#include <stout/os.hpp>
+
+#include "rendler_helper.hpp"
+#include "rendler_v1_executor.hpp"
+
+using std::cout;
+using std::endl;
+using std::vector;
+
+using mesos::vectorToString;
+
+static string renderJSPath;
+static string workDirPath;
+
+
+class RenderV1Executor : public RendlerV1Executor
+{
+public:
+  RenderV1Executor(const FrameworkID& _frameworkId,
+                   const ExecutorID& _executorId)
+    : RendlerV1Executor("RenderV1Executor", _frameworkId, _executorId) {}
+
+  virtual ~RenderV1Executor() {}
+
+protected:
+  void runTask(const TaskInfo& task) override
+  {
+    string url = task.data();
+    cout << "Running render task (" << task.task_id().value() << "): " << url;
+    string filename = workDirPath + task.task_id().value() + ".png";
+
+    vector<string> result;
+    result.push_back(task.task_id().value());
+    result.push_back(url);
+    result.push_back(filename);
+
+    string cmd = "QT_QPA_PLATFORM=offscreen phantomjs " + renderJSPath + " " + url + " " + filename;
+    assert(system(cmd.c_str()) != -1);
+
+    sendFrameworkMessage(vectorToString(result));
+    sendStatusUpdate(task, TASK_FINISHED);
+  }
+};
+
+
+int main(int argc, char** argv)
+{
+  FrameworkID frameworkId;
+  ExecutorID executorId;
+
+  Option<string> value;
+
+  value = os::getenv("MESOS_FRAMEWORK_ID");
+  if (value.isNone()) {
+    EXIT(EXIT_FAILURE)
+      << "Expecting 'MESOS_FRAMEWORK_ID' to be set in the environment";
+  }
+  frameworkId.set_value(value.get());
+
+  value = os::getenv("MESOS_EXECUTOR_ID");
+  if (value.isNone()) {
+    EXIT(EXIT_FAILURE)
+      << "Expecting 'MESOS_EXECUTOR_ID' to be set in the environment";
+  }
+  executorId.set_value(value.get());
+
+  std::string path = os::realpath(::dirname(argv[0])).get();
+  renderJSPath = path + "/render.js";
+  workDirPath = path + "/rendler-work-dir/";
+
+  process::Owned<RenderV1Executor> renderer(
+      new RenderV1Executor(frameworkId, executorId));
+
+  process::spawn(renderer.get());
+  process::wait(renderer.get());
+
+  return 0;
+}

cpp/rendler.cpp

@@ -96,6 +96,7 @@ class Rendler : public Scheduler
     for (size_t i = 0; i < offers.size(); i++) {
       const Offer& offer = offers[i];
       Resources remaining = offer.resources();
+      remaining.unallocate();
 
       static Resources TASK_RESOURCES = Resources::parse(
           "cpus:" + stringify<float>(CPUS_PER_TASK) +