http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/docs/deployment-configuration.md ---------------------------------------------------------------------- diff --git a/docs/deployment-configuration.md b/docs/deployment-configuration.md index d0ef37d..70b5500 100644 --- a/docs/deployment-configuration.md +++ b/docs/deployment-configuration.md @@ -29,7 +29,7 @@ gearpump{ ## Configuration for user submitted application job -For user application job, it will read configuration file `gear.conf` and `application.conf` from classpath, while `application.conf` has higher prioprty. +For user application job, it will read configuration file `gear.conf` and `application.conf` from classpath, while `application.conf` has higher priority. The default classpath contains: 1. `conf/` @@ -37,7 +37,6 @@ The default classpath contains: For example, you can put a `application.conf` on your working directory, and then it will be effective when you submit a new job application. - ## Logging To change the log level, you need to change both `gear.conf`, and `log4j.properties`.
http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/docs/deployment-msg-delivery.md ---------------------------------------------------------------------- diff --git a/docs/deployment-msg-delivery.md b/docs/deployment-msg-delivery.md index ff6de1f..10208ba 100644 --- a/docs/deployment-msg-delivery.md +++ b/docs/deployment-msg-delivery.md @@ -3,10 +3,9 @@ layout: global title: Deployment for Reliable Message Delivery --- - ## How to deploy for At Least Once Message Delivery? -As introduced in the [What is At Least Once Message Delievery](message-delivery.html#what-is-at-least-once-message-delivery), Gearpump has a built in KafkaSource. To get at least once message delivery, users should deploy a Kafka cluster as the offset store along with the Gearpump cluster. +As introduced in the [What is At Least Once Message Delivery](message-delivery.html#what-is-at-least-once-message-delivery), Gearpump has a built in KafkaSource. To get at least once message delivery, users should deploy a Kafka cluster as the offset store along with the Gearpump cluster. Here's an example to deploy a local Kafka cluster. @@ -31,7 +30,6 @@ Here's an example to deploy a local Kafka cluster. val source = new KafkaSource("topic1", "localhost:2181", offsetStorageFactory) ``` - ## How to deploy for Exactly Once Message Delivery? Exactly Once Message Delivery requires both an offset store and a checkpoint store. For the offset store, a Kafka cluster should be deployed as in the previous section. As for the checkpoint store, Gearpump has built-in support for Hadoop file systems, like HDFS. Hence, users should deploy a HDFS cluster alongside the Gearpump cluster. http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/docs/deployment-resource-isolation.md ---------------------------------------------------------------------- diff --git a/docs/deployment-resource-isolation.md b/docs/deployment-resource-isolation.md index 780dd60..0cb3764 100644 --- a/docs/deployment-resource-isolation.md +++ b/docs/deployment-resource-isolation.md @@ -9,7 +9,7 @@ cgroups (abbreviated from control groups) is a Linux kernel feature to limit, ac ## Start CGroup Service -Cgroups feature is only supported by Linux whose kernel version is larger than 2.6.18. Please also make sure the SELinux is disabled before start CGroup. +CGroup feature is only supported by Linux whose kernel version is larger than 2.6.18. Please also make sure the SELinux is disabled before start CGroup. The following steps are supposed to be executed by root user. @@ -30,7 +30,7 @@ The following steps are supposed to be executed by root user. perf_event /sys/fs/cgroup/perf_event ``` -3. If you want to assign permission to user **gear** to launch Gearpump Worker and applications with resouce isolation enabled, you need to check gear's uid and gid in /etc/passwd file, let's take **500** for example. +3. If you want to assign permission to user **gear** to launch Gearpump Worker and applications with resource isolation enabled, you need to check gear's uid and gid in /etc/passwd file, let's take **500** for example. 4. Add following content to /etc/cgconfig.conf @@ -71,7 +71,7 @@ The following steps are supposed to be executed by root user. 5. There should be a folder **gearpump** generated under the mount point of cpu subsystem and its owner is **gear:gear**. -6. Repeat the above-mentioned steps on each machine where you want to lauch Gearpump. +6. Repeat the above-mentioned steps on each machine where you want to launch Gearpump. ## Enable Cgroups in Gearpump 1. Login into the machine which has CGroup prepared with user **gear**. @@ -90,11 +90,11 @@ The following steps are supposed to be executed by root user. Please note the gearpump.cgroup.root **gearpump** must be consistent with the group name in /etc/cgconfig.conf. -3. Repeat the above-mentioned steps on each machine where you want to lauch Gearpump +3. Repeat the above-mentioned steps on each machine where you want to launch Gearpump 4. Start the Gearpump cluster, please refer to [Deploy Gearpump in Standalone Mode](deployment-standalone.html) -## Launch Application From Commad Line +## Launch Application From Command Line 1. Login into the machine which has Gearpump distribution. 2. Enter into Gearpump's home folder, edit gear.conf under folder ```${GEARPUMP_HOME}/conf/``` @@ -117,4 +117,4 @@ The following steps are supposed to be executed by root user. If you want to submit the application from dashboard, by default the ```gearpump.cgroup.cpu-core-limit-per-executor``` is inherited from Worker's configuration. You can provide your own conf file to override it. ## Limitations -Windows and Mac OS X don't support CGroup, so the resource isolation will not work even if you turn it on. There will not be any limitition for single executor's cpu usage. \ No newline at end of file +Windows and Mac OS X don't support CGroup, so the resource isolation will not work even if you turn it on. There will not be any limitation for single executor's cpu usage. \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/docs/deployment-security.md ---------------------------------------------------------------------- diff --git a/docs/deployment-security.md b/docs/deployment-security.md index 4be1346..9ce3715 100644 --- a/docs/deployment-security.md +++ b/docs/deployment-security.md @@ -36,7 +36,7 @@ Suppose user `gear` will launch gearpump on YARN, then the corresponding princip Before calling "yarnclient launch", make sure you have put all yarn configuration files under classpath. Typically, you can just copy all files under `$HADOOP_HOME/etc/hadoop` from one of the YARN cluster machine to `conf/yarnconf` of gearpump. `$HADOOP_HOME` points to the Hadoop installation directory. -5. Get Kerberos credentials to sumbit the job: +5. Get Kerberos credentials to submit the job: ``` kinit gearpump/[email protected] @@ -79,7 +79,7 @@ Since Gearpumpâs Master-Worker structure is similar to HDFSâs NameNode-DataN 1. User creates kerberos principal and keytab for Gearpump. 2. Deploy the keytab files to all the cluster nodes. -3. Configure Gearpumpâs conf file, specify kerberos principal and local keytab file localtion. +3. Configure Gearpumpâs conf file, specify kerberos principal and local keytab file location. 4. Start Master and Worker. Every application has a submitter/user. We will separate the application from different users, like different log folders for different applications. http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/docs/deployment-standalone.md ---------------------------------------------------------------------- diff --git a/docs/deployment-standalone.md b/docs/deployment-standalone.md index 06e701c..d77ffc8 100644 --- a/docs/deployment-standalone.md +++ b/docs/deployment-standalone.md @@ -9,7 +9,6 @@ Standalone mode is a distributed cluster mode. That is, Gearpump runs as service To deploy Gearpump in cluster mode, please first check that the [Pre-requisites](hardware-requirement.html) are met. - ### How to Install You need to have Gearpump binary at hand. Please refer to [How to get gearpump distribution](get-gearpump-distribution.html) to get the Gearpump binary. http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/docs/deployment-ui-authentication.md ---------------------------------------------------------------------- diff --git a/docs/deployment-ui-authentication.md b/docs/deployment-ui-authentication.md index 04a9c7b..872a857 100644 --- a/docs/deployment-ui-authentication.md +++ b/docs/deployment-ui-authentication.md @@ -5,7 +5,6 @@ title: UI Dashboard Authentication and Authorization ## What is this about? - ## How to enable UI authentication? 1. Change config file gear.conf, find entry `gearpump-ui.gearpump.ui-security.authentication-enabled`, change the value to true @@ -48,7 +47,6 @@ is the steps to configure ConfigFileBasedAuthenticator. For the default authentication plugin, it has three categories of users: admins, users, and guests. - * admins: have unlimited permission, like shutdown a cluster, add/remove machines. * users: have limited permission to submit an application and etc.. * guests: can not submit/kill applications, but can view the application status. @@ -245,7 +243,7 @@ To use Google OAuth2 Authenticator, there are several steps: 3. Config gear.conf `gearpump.ui-security.oauth2-authenticators.cloudfoundryuaa` section. Please make sure class name, client ID, client Secret, and callback URL are set properly. -**NOTE:** The callback URL here should matche what you set on CloudFoundry UAA in step1. +**NOTE:** The callback URL here should match what you set on CloudFoundry UAA in step1. #### Step3: Configure network proxy for Gearpump UI server if applies @@ -286,7 +284,7 @@ You can follow the Google OAuth2 example code to define a custom OAuth2Authentic * parameters and form fields. * * @note '''Thread-safety''' is a MUST requirement. Developer need to ensure the sub-class is thread-safe. - * Sub-class should have a parameterless constructor. + * Sub-class should have a parameter-less constructor. * * @note OAuth2 Authenticator requires access of Internet. Please make sure HTTP proxy are * set properly if applied. @@ -374,4 +372,3 @@ You can follow the Google OAuth2 example code to define a custom OAuth2Authentic ``` The configuration entry is supposed to be used by class `SocialNetworkXAuthenticator`. - http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/docs/dev-connectors.md ---------------------------------------------------------------------- diff --git a/docs/dev-connectors.md b/docs/dev-connectors.md index a9983fc..08ff023 100644 --- a/docs/dev-connectors.md +++ b/docs/dev-connectors.md @@ -9,7 +9,7 @@ title: Gearpump Connectors ### DataSource `DataSource` is the concept in Gearpump that without input and will output messages. So, basically, `DataSource` is the start point of a streaming processing flow. -As Gearpump depends on `DataSource` to be replayable to ensure at-least-once message delivery and exactly-once message delivery, for some data sources, we will need a `io.gearpump.streaming.transaction.api.OffsetStorageFactory` to store the offset (progress) of current `DataSource`. So that, when a replay is needed, Gearpump can guide `DataSource` to replay from certain offset. +As Gearpump depends on `DataSource` to be replay-able to ensure at-least-once message delivery and exactly-once message delivery, for some data sources, we will need a `io.gearpump.streaming.transaction.api.OffsetStorageFactory` to store the offset (progress) of current `DataSource`. So that, when a replay is needed, Gearpump can guide `DataSource` to replay from certain offset. Currently Gearpump `DataSource` only support infinite stream. Finite stream support will be added in a near future release. @@ -34,7 +34,6 @@ Name | Description `HBaseSink` | Write the message to HBase. The message to write must be HBase `Put` or a tuple of `(rowKey, family, column, value)`. `KafkaSink` | Write to Kafka. - ## Use of Connectors ### Use of `KafkaSource` @@ -88,7 +87,6 @@ Then, you can use `KafkaSource` in your application: To use `HBaseSink` in your application, you first need to add the `gearpump-external-hbase` library dependency in your application: - ``` "com.github.intel-hadoop" %% "gearpump-external-hbase" % {{ site.GEARPUMP_VERSION }} ``` @@ -101,7 +99,6 @@ To use `HBaseSink` in your application, you first need to add the `gearpump-exte </dependency> ``` - To connect to HBase, you need to provide following info: - the HBase configuration to tell which HBase service to connect - the table name (you must create the table yourself, see the [HBase documentation](https://hbase.apache.org/book.html)) http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/docs/dev-custom-serializer.md ---------------------------------------------------------------------- diff --git a/docs/dev-custom-serializer.md b/docs/dev-custom-serializer.md index ca0a332..5be0544 100644 --- a/docs/dev-custom-serializer.md +++ b/docs/dev-custom-serializer.md @@ -41,7 +41,6 @@ but rather import io.gearpump.google.common.io.Files ``` - ##### System Level Serializer If the serializer is widely used, you can define a global serializer which is available to all applications(or worker or master) in the system. @@ -128,7 +127,6 @@ gearpump { ###### Step3: All set! - #### Advanced: Choose another serialization framework Note: This is only for advanced user which require deep customization of Gearpump platform. http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/docs/dev-non-streaming-example.md ---------------------------------------------------------------------- diff --git a/docs/dev-non-streaming-example.md b/docs/dev-non-streaming-example.md index eac92d0..1aec2d1 100644 --- a/docs/dev-non-streaming-example.md +++ b/docs/dev-non-streaming-example.md @@ -11,7 +11,6 @@ What Distributed Shell do is that user send a shell command to the cluster and t Repository and library dependencies can be found at [Maven Setting](maven-setting.html) - ### Define Executor Class ```scala http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/docs/dev-rest-api.md ---------------------------------------------------------------------- diff --git a/docs/dev-rest-api.md b/docs/dev-rest-api.md index 7b2bb6f..6243a89 100644 --- a/docs/dev-rest-api.md +++ b/docs/dev-rest-api.md @@ -579,8 +579,6 @@ Sample Response: {"enabled":true} ``` - - ### GET api/v1.0/supervisor Get the supervisor path @@ -616,7 +614,6 @@ Sample Response: ### POST api/v1.0/supervisor/removeworker/<worker-id> Remove single worker instance by specifying a worker Id. - **NOTE:* Use with caution! **NOTE:** All executors JVMs under this worker JVM will also be destroyed. It will trigger failover for all @@ -637,7 +634,6 @@ Sample Response: ## Application service - ### GET api/v1.0/appmaster/<appId>?detail=<true|false> Query information of an specific application of Id appId @@ -793,7 +789,6 @@ Sample Response: } ``` - ### GET api/v1.0/appmaster/<appId>/config Query the configuration of specific application appId http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/docs/dev-storm.md ---------------------------------------------------------------------- diff --git a/docs/dev-storm.md b/docs/dev-storm.md index 914aabe..129f38e 100644 --- a/docs/dev-storm.md +++ b/docs/dev-storm.md @@ -4,7 +4,7 @@ title: Storm Compatibility --- Gearpump provides **binary compatibility** for Apache Storm applications. That is to say, users could easily grab an existing Storm jar and run it -on Gearpump. This documentation illustrates Gearpump's comapatibility with Storm. +on Gearpump. This documentation illustrates Gearpump's compatibility with Storm. ## What Storm features are supported on Gearpump @@ -32,7 +32,7 @@ on Gearpump. This documentation illustrates Gearpump's comapatibility with Storm | storm-jdbc | yes | | storm-redis | yes | | flux | yes | -| storm-eventhubs | not verfied | +| storm-eventhubs | not verified | | Trident | no | ### At Least Once support @@ -95,7 +95,6 @@ This section shows how to run an existing Storm jar in a local Gearpump cluster. a. submit Storm applications through command line - ``` bin/storm app -verbose -config app.yaml -jar storm-starter-${STORM_VERSION}.jar storm.starter.ExclamationTopology exclamation ``` @@ -138,14 +137,14 @@ Here's an example of `WordCountTopology` with acker bolts (ackers) being transla Gearpump creates a `StormProducer` for each Storm spout and a `StormProcessor` for each Storm bolt (except for ackers) with the same parallelism, and wires them together using the same grouping strategy (partitioning in Gearpump) as in Storm. -At runtime, spouts and bolts are running inside `StormProducer` tasks and `StormProcessor` tasks respectively. Messages emitted by spout are passed to `StormProducer`, transferred to `StormProcessor` and passed down to bolt. Messages are serialized / deserialized with Storm serializers. +At runtime, spouts and bolts are running inside `StormProducer` tasks and `StormProcessor` tasks respectively. Messages emitted by spout are passed to `StormProducer`, transferred to `StormProcessor` and passed down to bolt. Messages are serialized / de-serialized with Storm serializers. Storm ackers are dropped since Gearpump has a different mechanism of message tracking and flow control. ### Task execution Each Storm task is executed by a dedicated thread while all Gearpump tasks of an executor share a thread pool. Generally, we can achieve better performance with a shared thread pool. It's possible, however, some tasks block and take up all the threads. In that case, we can -fall back to the Storm way by setting `gearpump.task-dispatcher` to `"gaerpump.single-thread-dispatcher"` in `gear.conf`. +fall back to the Storm way by setting `gearpump.task-dispatcher` to `"gearpump.single-thread-dispatcher"` in `gear.conf`. ### Message tracking @@ -153,7 +152,6 @@ Storm tracks the lineage of each message with ackers to guarantee at-least-once Gearpump [tracks messages between a sender and receiver in an efficient way](gearpump-internals.html#how-do-we-detect-message-loss). Message loss causes the whole application to replay from the [minimum timestamp of all pending messages in the system](gearpump-internals.html#application-clock-and-global-clock-service). - ### Flow control Storm throttles flow rate at spout, which stops sending messages if the number of unacked messages exceeds `topology.max.spout.pending`. @@ -185,7 +183,6 @@ Since StreamCQL already supports Storm, it's straightforward to run StreamCQL ov 3. Go to the installed stream-cql-binary, and change following settings in `conf/streaming-site.xml` with the output Nimbus configs in Step 2. - ```xml <property> <name>streaming.storm.nimbus.host</name> http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/docs/dev-write-1st-app.md ---------------------------------------------------------------------- diff --git a/docs/dev-write-1st-app.md b/docs/dev-write-1st-app.md index 738790b..1dd47b6 100644 --- a/docs/dev-write-1st-app.md +++ b/docs/dev-write-1st-app.md @@ -55,8 +55,6 @@ object Split { } ``` - - Like Split, every processor extends a `TaskActor`. The `onStart` method is called once before any message comes in; `onNext` method is called to process every incoming message. Note that Gearpump employs the message-driven model and that's why Split sends itself a message at the end of `onStart` and `onNext` to trigger next message processing. #### Sum Processor @@ -157,13 +155,12 @@ object WordCount extends App with ArgumentsParser { We override `options` value and define an array of command line arguments to parse. We want application users to pass in masters' hosts and ports, the parallelism of split and sum tasks, and how long to run the example. We also specify whether an option is `required` and provide `defaultValue` for some arguments. -Given the `ParseResult` of command line arguments, we create `TaskDescription`s for Split and Sum processors, and connect them with `HashPartitioner` using DAG API. The graph is wrapped in an `AppDescrition` , which is finally submit to master. +Given the `ParseResult` of command line arguments, we create `TaskDescription`s for Split and Sum processors, and connect them with `HashPartitioner` using DAG API. The graph is wrapped in an `AppDescription` , which is finally submit to master. ### Submit application After all these, you need to package everything into a uber jar and submit the jar to Gearpump Cluster. Please check [Application submission tool](commandline.html) to command line tool syntax. - ### Advanced topic For a real application, you definitely need to define your own customized message passing between processors. Customized message needs customized serializer to help message passing over wire. http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/docs/features.md ---------------------------------------------------------------------- diff --git a/docs/features.md b/docs/features.md index 2bd9417..210ac7a 100644 --- a/docs/features.md +++ b/docs/features.md @@ -7,8 +7,7 @@ description: Gearpump Technical Highlights ### Technical highlights of Gearpump - -Gearpump is a performant, flexible, fault-tolerant, and responsive streaming platform with a lot of nice features, its technical highlights include: +Gearpump is a high performance, flexible, fault-tolerant, and responsive streaming platform with a lot of nice features, its technical highlights include: #### Actors everywhere http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/docs/gearpump-internals.md ---------------------------------------------------------------------- diff --git a/docs/gearpump-internals.md b/docs/gearpump-internals.md index 9221157..06e045d 100644 --- a/docs/gearpump-internals.md +++ b/docs/gearpump-internals.md @@ -4,7 +4,7 @@ displayTitle: Gearpump Internals title: Gearpump Internals description: Gearpump Internals --- -### Actor Hiearachy? +### Actor Hierarchy?  @@ -72,7 +72,7 @@ Without flow control, one task can easily flood another task with too many messa Figure: Flow control, each task is "star" connected to input tasks and output tasks The difficult part for our problem is that each task can have multiple input tasks and output tasks. The input and output must be geared together so that the back pressure can be properly propagated from downstream to upstream. The flow control also needs to consider failures, and it needs to be able to recover when there is message loss. -Another challenge is that the overhead of flow control messages can be big. If we ack every message, there will be huge amount of ack'd messages in the system, degrading streaming performance. The approach we adopted is to use explicit AckRequest message. The target tasks will only ack back when they receive the AckRequest message, and the source will only send AckRequest when it feels necessary. With this approach, we can largely reduce the overhead. +Another challenge is that the overhead of flow control messages can be big. If we ack every message, there will be huge amount of acked messages in the system, degrading streaming performance. The approach we adopted is to use explicit AckRequest message. The target tasks will only ack back when they receive the AckRequest message, and the source will only send AckRequest when it feels necessary. With this approach, we can largely reduce the overhead. ### How do we detect message loss? @@ -231,4 +231,4 @@ When there is message loss, the AppMaster will first pause the global clock serv Kafka queue only expose the offset information for each partition. What KafkaSource do is to maintain its own mapping from Kafka offset to Application timestamp, so that we can map from a application timestamp to a Kafka offset, and replay Kafka messages from that Kafka offset. -The mapping between Application timestmap with Kafka offset is stored in a distributed file system or as a Kafka topic. +The mapping between Application timestamp with Kafka offset is stored in a distributed file system or as a Kafka topic. http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/docs/hardware-requirement.md ---------------------------------------------------------------------- diff --git a/docs/hardware-requirement.md b/docs/hardware-requirement.md index 87e87c5..42e839c 100644 --- a/docs/hardware-requirement.md +++ b/docs/hardware-requirement.md @@ -25,7 +25,6 @@ CPU | Nothing special HDFS installation | Default is not required. You only need to install it when you want to store the application jars in HDFS. Kafka installation | Default is not required. You need to install Kafka when you want the at-least once message delivery feature. Currently, the only supported data source for this feature is Kafka - ** Table: The default port used in Gearpump:** | usage | Port | Description | http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/docs/how-to-contribute.md ---------------------------------------------------------------------- diff --git a/docs/how-to-contribute.md b/docs/how-to-contribute.md index 4c15aaf..e9e33c1 100644 --- a/docs/how-to-contribute.md +++ b/docs/how-to-contribute.md @@ -3,7 +3,6 @@ layout: global title: How to contribute --- - ## Contributions Welcome! Gearpump is developed by an open and friendly community. Everybody is cordially welcome to join the community and contribute to Gearpump. There are several ways to interact with the community and to contribute to Gearpump including asking questions, filing bug reports, implementing new use cases, proposing new features, joining discussions on the mailing lists, contributing code or documentation, improving the website, or testing release candidates. @@ -54,7 +53,6 @@ We welcome any contribution to improve our website. Please open an issue at [Gearpump Website Issue Tracker](https://github.com/gearpump/gearpump.github.io/issues) if you think our website could be improved. - ### More ways to contribute⦠There are many more ways to contribute to the Gearpump community. For example you can http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/docs/index.md ---------------------------------------------------------------------- diff --git a/docs/index.md b/docs/index.md index 75572c5..d24f3e4 100644 --- a/docs/index.md +++ b/docs/index.md @@ -27,12 +27,11 @@ Gearpump's feature set includes: * Samoa compatibility * Both high level and low level API - ### Gearpump Performance Per initial benchmarks we are able to process 18 million messages/second (100 bytes per message) with a 8ms latency on a 4-node cluster.  ### Gearpump and Akka -Gearump is a 100% Akka based platform. We model big data streaming within the Akka actor hierarchy. +Gearpump is a 100% Akka based platform. We model big data streaming within the Akka actor hierarchy.  http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/docs/js/api-docs.js ---------------------------------------------------------------------- diff --git a/docs/js/api-docs.js b/docs/js/api-docs.js index ce89d89..864502c 100644 --- a/docs/js/api-docs.js +++ b/docs/js/api-docs.js @@ -17,7 +17,7 @@ /* Dynamically injected post-processing code for the API docs */ -$(document).ready(function() { +$(document).ready(function () { var annotations = $("dt:contains('Annotations')").next("dd").children("span.name"); addBadges(annotations, "AlphaComponent", ":: AlphaComponent ::", '<span class="alphaComponent badge">Alpha Component</span>'); addBadges(annotations, "DeveloperApi", ":: DeveloperApi ::", '<span class="developer badge">Developer API</span>'); @@ -29,7 +29,7 @@ function addBadges(allAnnotations, name, tag, html) { var tags = $(".cmt:contains(" + tag + ")") // Remove identifier tags from comments - tags.each(function(index) { + tags.each(function (index) { var oldHTML = $(this).html(); var newHTML = oldHTML.replace(tag, ""); $(this).html(newHTML); http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/docs/js/api-javadocs.js ---------------------------------------------------------------------- diff --git a/docs/js/api-javadocs.js b/docs/js/api-javadocs.js index ead13d6..c4eb8a6 100644 --- a/docs/js/api-javadocs.js +++ b/docs/js/api-javadocs.js @@ -17,7 +17,7 @@ /* Dynamically injected post-processing code for the API docs */ -$(document).ready(function() { +$(document).ready(function () { addBadges(":: AlphaComponent ::", '<span class="alphaComponent badge">Alpha Component</span>'); addBadges(":: DeveloperApi ::", '<span class="developer badge">Developer API</span>'); addBadges(":: Experimental ::", '<span class="experimental badge">Experimental</span>'); @@ -27,24 +27,24 @@ function addBadges(tag, html) { var tags = $(".block:contains(" + tag + ")") // Remove identifier tags - tags.each(function(index) { + tags.each(function (index) { var oldHTML = $(this).html(); var newHTML = oldHTML.replace(tag, ""); $(this).html(newHTML); }); // Add html badge tags - tags.each(function(index) { + tags.each(function (index) { if ($(this).parent().is('td.colLast')) { $(this).parent().prepend(html); } else if ($(this).parent('li.blockList') - .parent('ul.blockList') - .parent('div.description') - .parent().is('div.contentContainer')) { + .parent('ul.blockList') + .parent('div.description') + .parent().is('div.contentContainer')) { var contentContainer = $(this).parent('li.blockList') - .parent('ul.blockList') - .parent('div.description') - .parent('div.contentContainer') + .parent('ul.blockList') + .parent('div.description') + .parent('div.contentContainer') var header = contentContainer.prev('div.header'); if (header.length > 0) { header.prepend(html); http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/docs/js/main.js ---------------------------------------------------------------------- diff --git a/docs/js/main.js b/docs/js/main.js index 0bac247..ae989af 100644 --- a/docs/js/main.js +++ b/docs/js/main.js @@ -25,7 +25,7 @@ function codeTabs() { "python": "img/python-sm.png", "java": "img/java-sm.png" }; - $("div.codetabs").each(function() { + $("div.codetabs").each(function () { $(this).addClass("tab-content"); // Insert the tab bar @@ -34,7 +34,7 @@ function codeTabs() { // Add each code sample to the tab bar: var codeSamples = $(this).children("div"); - codeSamples.each(function() { + codeSamples.each(function () { $(this).addClass("tab-pane"); var lang = $(this).data("lang"); var image = $(this).data("image"); @@ -44,7 +44,7 @@ function codeTabs() { var id = "tab_" + lang + "_" + counter; $(this).attr("id", id); if (image != null && langImages[lang]) { - var buttonLabel = "<img src='" +langImages[lang] + "' alt='" + capitalizedLang + "' />"; + var buttonLabel = "<img src='" + langImages[lang] + "' alt='" + capitalizedLang + "' />"; } else if (notabs == null) { var buttonLabel = "<b>" + capitalizedLang + "</b>"; } else { @@ -67,12 +67,11 @@ function codeTabs() { $(this).tab('show'); $(document).scrollTop($(this).offset().top - scrollOffset); }); - $("table").each(function() { + $("table").each(function () { $(this).addClass("table table-bordered"); }); } - // A script to fix internal hash links because we have an overlapping top bar. // Based on https://github.com/twitter/bootstrap/issues/193#issuecomment-2281510 function maybeScrollToHash() { @@ -82,7 +81,7 @@ function maybeScrollToHash() { } } -$(function() { +$(function () { codeTabs(); // Display anchor links when hovering over headers. For documentation of the // configuration options, see the AnchorJS documentation. @@ -91,11 +90,15 @@ $(function() { }; anchors.add(); - $(window).bind('hashchange', function() { + $(window).bind('hashchange', function () { maybeScrollToHash(); }); // Scroll now too in case we had opened the page on a hash, but wait a bit because some browsers // will try to do *their* initial scroll after running the onReady handler. - $(window).load(function() { setTimeout(function() { maybeScrollToHash(); }, 25); }); + $(window).load(function () { + setTimeout(function () { + maybeScrollToHash(); + }, 25); + }); }); http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/docs/message-delivery.md ---------------------------------------------------------------------- diff --git a/docs/message-delivery.md b/docs/message-delivery.md index 781bfd5..909574f 100644 --- a/docs/message-delivery.md +++ b/docs/message-delivery.md @@ -3,16 +3,12 @@ layout: global title: Reliable Message Delivery --- - - ## What is At Least Once Message Delivery? -Messages could be lost on delivery due to network partitions. **At Least Once Message Delivery** (at least once) means the lost messages are delivered one or more times such that at least one is processed and acked by the whole flow. - -Gearpump guarantees at least once for any source that is able to replay message from a past timestamp. In Gearpump, each message is tagged with a timestamp, and the system tracks the minimum timestamp of all pending messages (the global minimum clock). On message loss, application will be restarted to the global minimum clock. Since the source is able to replay from the gloabl minimum clock, all pending messages before the restart will be replayed. Gearpump calls that kind of source `TimeReplayableSource` and already provides a built in -[KafkaSource](gearpump-internals.html#at-least-once-message-delivery-and-kafka). With the KafkaSource to ingest data into Gearpump, users are guaranteed at least once message delievery. - +Messages could be lost on delivery due to network partitions. **At Least Once Message Delivery** (at least once) means the lost messages are delivered one or more times such that at least one is processed and acknowledged by the whole flow. +Gearpump guarantees at least once for any source that is able to replay message from a past timestamp. In Gearpump, each message is tagged with a timestamp, and the system tracks the minimum timestamp of all pending messages (the global minimum clock). On message loss, application will be restarted to the global minimum clock. Since the source is able to replay from the global minimum clock, all pending messages before the restart will be replayed. Gearpump calls that kind of source `TimeReplayableSource` and already provides a built in +[KafkaSource](gearpump-internals.html#at-least-once-message-delivery-and-kafka). With the KafkaSource to ingest data into Gearpump, users are guaranteed at least once message delivery. ## What is Exactly Once Message Delivery? @@ -24,7 +20,6 @@ Users are guaranteed exactly once in Gearpump if they use both a `TimeReplayable ### Persistent API Persistent API consists of `PersistentTask` and `PersistentState`. - Here is an example of using them to keep count of incoming messages. ```scala http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/examples/distributedshell/src/main/scala/io/gearpump/examples/distributedshell/DistShellAppMaster.scala ---------------------------------------------------------------------- diff --git a/examples/distributedshell/src/main/scala/io/gearpump/examples/distributedshell/DistShellAppMaster.scala b/examples/distributedshell/src/main/scala/io/gearpump/examples/distributedshell/DistShellAppMaster.scala index 8533a03..9b417c0 100644 --- a/examples/distributedshell/src/main/scala/io/gearpump/examples/distributedshell/DistShellAppMaster.scala +++ b/examples/distributedshell/src/main/scala/io/gearpump/examples/distributedshell/DistShellAppMaster.scala @@ -7,7 +7,7 @@ * "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 + * 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, @@ -17,20 +17,23 @@ */ package io.gearpump.examples.distributedshell +import scala.concurrent.Future + import akka.actor.{Deploy, Props} import akka.pattern.{ask, pipe} import akka.remote.RemoteScope import com.typesafe.config.Config +import org.slf4j.Logger + import io.gearpump.cluster.ClientToMaster.ShutdownApplication import io.gearpump.cluster.appmaster.ExecutorSystemScheduler.{ExecutorSystemJvmConfig, ExecutorSystemStarted, StartExecutorSystemTimeout} import io.gearpump.cluster.{AppDescription, AppMasterContext, ApplicationMaster, ExecutorContext} -import DistShellAppMaster._ +import io.gearpump.examples.distributedshell.DistShellAppMaster._ import io.gearpump.util.{ActorUtil, Constants, LogUtil, Util} -import org.slf4j.Logger -import scala.concurrent.Future +class DistShellAppMaster(appContext: AppMasterContext, app: AppDescription) + extends ApplicationMaster { -class DistShellAppMaster(appContext : AppMasterContext, app : AppDescription) extends ApplicationMaster { import appContext._ import context.dispatcher implicit val timeout = Constants.FUTURE_TIMEOUT @@ -45,10 +48,11 @@ class DistShellAppMaster(appContext : AppMasterContext, app : AppDescription) ex override def receive: Receive = { case ExecutorSystemStarted(executorSystem, _) => import executorSystem.{address, resource => executorResource, worker} - val executorContext = ExecutorContext(currentExecutorId, worker, appId, app.name, self, executorResource) - //start executor + val executorContext = ExecutorContext(currentExecutorId, worker, appId, app.name, + self, executorResource) + // Start executor val executor = context.actorOf(Props(classOf[ShellExecutor], executorContext, app.userConfig) - .withDeploy(Deploy(scope = RemoteScope(address))), currentExecutorId.toString) + .withDeploy(Deploy(scope = RemoteScope(address))), currentExecutorId.toString) executorSystem.bindLifeCycleWith(executor) currentExecutorId += 1 case StartExecutorSystemTimeout => @@ -56,14 +60,17 @@ class DistShellAppMaster(appContext : AppMasterContext, app : AppDescription) ex masterProxy ! ShutdownApplication(appId) context.stop(self) case msg: ShellCommand => - Future.fold(context.children.map(_ ? msg))(new ShellCommandResultAggregator) { (aggregator, response) => - aggregator.aggregate(response.asInstanceOf[ShellCommandResult]) + Future.fold(context.children.map(_ ? msg))(new ShellCommandResultAggregator) { + (aggregator, response) => { + aggregator.aggregate(response.asInstanceOf[ShellCommandResult]) + } }.map(_.toString()) pipeTo sender } private def getExecutorJvmConfig: ExecutorSystemJvmConfig = { val config: Config = app.clusterConfig - val jvmSetting = Util.resolveJvmSetting(config.withFallback(context.system.settings.config)).executor + val jvmSetting = Util.resolveJvmSetting(config.withFallback(context.system.settings.config)) + .executor ExecutorSystemJvmConfig(jvmSetting.classPath, jvmSetting.vmargs, appJar, username, config) } @@ -83,6 +90,6 @@ object DistShellAppMaster { this } - override def toString() = result.toString() + override def toString(): String = result.toString() } } \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/examples/distributedshell/src/main/scala/io/gearpump/examples/distributedshell/DistributedShell.scala ---------------------------------------------------------------------- diff --git a/examples/distributedshell/src/main/scala/io/gearpump/examples/distributedshell/DistributedShell.scala b/examples/distributedshell/src/main/scala/io/gearpump/examples/distributedshell/DistributedShell.scala index ac47740..b7b3eb0 100644 --- a/examples/distributedshell/src/main/scala/io/gearpump/examples/distributedshell/DistributedShell.scala +++ b/examples/distributedshell/src/main/scala/io/gearpump/examples/distributedshell/DistributedShell.scala @@ -7,7 +7,7 @@ * "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 + * 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, @@ -17,12 +17,14 @@ */ package io.gearpump.examples.distributedshell -import io.gearpump.cluster.{Application, UserConfig, AppDescription} +import org.slf4j.Logger + import io.gearpump.cluster.client.ClientContext -import io.gearpump.cluster.main.{ParseResult, CLIOption, ArgumentsParser} +import io.gearpump.cluster.main.{ArgumentsParser, CLIOption} +import io.gearpump.cluster.{Application, UserConfig} import io.gearpump.util.{AkkaApp, LogUtil} -import org.slf4j.Logger +/** Demo application to distribute and execute shell command on all machines of the cluster */ object DistributedShell extends AkkaApp with ArgumentsParser { private val LOG: Logger = LogUtil.getLogger(getClass) @@ -31,7 +33,8 @@ object DistributedShell extends AkkaApp with ArgumentsParser { override def main(akkaConf: Config, args: Array[String]): Unit = { LOG.info(s"Distributed shell submitting application...") val context = ClientContext(akkaConf) - val appId = context.submit(Application[DistShellAppMaster]("DistributedShell", UserConfig.empty)) + val appId = context.submit(Application[DistShellAppMaster]("DistributedShell", + UserConfig.empty)) context.close() LOG.info(s"Distributed Shell Application started with appId $appId !") } http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/examples/distributedshell/src/main/scala/io/gearpump/examples/distributedshell/DistributedShellClient.scala ---------------------------------------------------------------------- diff --git a/examples/distributedshell/src/main/scala/io/gearpump/examples/distributedshell/DistributedShellClient.scala b/examples/distributedshell/src/main/scala/io/gearpump/examples/distributedshell/DistributedShellClient.scala index feecd18..cd6f943 100644 --- a/examples/distributedshell/src/main/scala/io/gearpump/examples/distributedshell/DistributedShellClient.scala +++ b/examples/distributedshell/src/main/scala/io/gearpump/examples/distributedshell/DistributedShellClient.scala @@ -7,7 +7,7 @@ * "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 + * 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, @@ -18,19 +18,19 @@ package io.gearpump.examples.distributedshell import java.util.concurrent.TimeUnit +import scala.concurrent.Await +import scala.concurrent.duration.Duration + +import akka.pattern.ask +import org.slf4j.{Logger, LoggerFactory} import io.gearpump.cluster.client.ClientContext import io.gearpump.cluster.main.{ArgumentsParser, CLIOption} -import DistShellAppMaster.ShellCommand - -import akka.pattern.ask +import io.gearpump.examples.distributedshell.DistShellAppMaster.ShellCommand import io.gearpump.util.{AkkaApp, Constants} -import org.slf4j.{LoggerFactory, Logger} - -import scala.concurrent.Await -import scala.concurrent.duration.Duration -object DistributedShellClient extends AkkaApp with ArgumentsParser { +/** Client to DistributedShell to input "shell command" */ +object DistributedShellClient extends AkkaApp with ArgumentsParser { implicit val timeout = Constants.FUTURE_TIMEOUT private val LOG: Logger = LoggerFactory.getLogger(getClass) http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/examples/distributedshell/src/main/scala/io/gearpump/examples/distributedshell/ShellExecutor.scala ---------------------------------------------------------------------- diff --git a/examples/distributedshell/src/main/scala/io/gearpump/examples/distributedshell/ShellExecutor.scala b/examples/distributedshell/src/main/scala/io/gearpump/examples/distributedshell/ShellExecutor.scala index 712788c..2d0fd06 100644 --- a/examples/distributedshell/src/main/scala/io/gearpump/examples/distributedshell/ShellExecutor.scala +++ b/examples/distributedshell/src/main/scala/io/gearpump/examples/distributedshell/ShellExecutor.scala @@ -7,7 +7,7 @@ * "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 + * 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, @@ -18,16 +18,18 @@ package io.gearpump.examples.distributedshell +import scala.sys.process._ +import scala.util.{Failure, Success, Try} + import akka.actor.Actor -import io.gearpump.cluster.{UserConfig, ExecutorContext} -import DistShellAppMaster.{ShellCommandResult, ShellCommand} -import io.gearpump.util.LogUtil import org.slf4j.Logger -import scala.util.{Failure, Success, Try} -import sys.process._ +import io.gearpump.cluster.{ExecutorContext, UserConfig} +import io.gearpump.examples.distributedshell.DistShellAppMaster.{ShellCommand, ShellCommandResult} +import io.gearpump.util.LogUtil -class ShellExecutor(executorContext: ExecutorContext, userConf : UserConfig) extends Actor{ +/** Executor actor on remote machine */ +class ShellExecutor(executorContext: ExecutorContext, userConf: UserConfig) extends Actor { import executorContext._ private val LOG: Logger = LogUtil.getLogger(getClass, executor = executorId, app = appId) @@ -35,7 +37,7 @@ class ShellExecutor(executorContext: ExecutorContext, userConf : UserConfig) ext override def receive: Receive = { case ShellCommand(command) => - val process = Try(s"$command" !!) + val process = Try(s"$command".!!) val result = process match { case Success(msg) => msg case Failure(ex) => ex.getMessage http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/DistShellAppMasterSpec.scala ---------------------------------------------------------------------- diff --git a/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/DistShellAppMasterSpec.scala b/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/DistShellAppMasterSpec.scala index ec6bba1..2d63734 100644 --- a/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/DistShellAppMasterSpec.scala +++ b/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/DistShellAppMasterSpec.scala @@ -7,7 +7,7 @@ * "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 + * 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, @@ -17,20 +17,24 @@ */ package io.gearpump.examples.distributedshell +import scala.concurrent.Await +import scala.concurrent.duration.Duration + import akka.actor.ActorSystem import akka.testkit.{TestActorRef, TestProbe} -import io.gearpump.WorkerId -import io.gearpump.cluster.AppMasterToMaster.{RequestResource, GetAllWorkers, RegisterAppMaster} +import org.scalatest.{BeforeAndAfter, Matchers, WordSpec} + +import io.gearpump.cluster.AppMasterToMaster.{GetAllWorkers, RegisterAppMaster, RequestResource} import io.gearpump.cluster.AppMasterToWorker.LaunchExecutor -import io.gearpump.cluster.MasterToAppMaster.{ResourceAllocated, WorkerList, AppMasterRegistered} +import io.gearpump.cluster.MasterToAppMaster.{AppMasterRegistered, ResourceAllocated, WorkerList} import io.gearpump.cluster._ -import io.gearpump.cluster.appmaster.{AppMasterRuntimeInfo, AppMasterRuntimeEnvironment} -import io.gearpump.cluster.scheduler.{ResourceAllocation, Relaxation, ResourceRequest, Resource} +import io.gearpump.cluster.appmaster.{AppMasterRuntimeEnvironment, AppMasterRuntimeInfo} +import io.gearpump.cluster.scheduler.{Relaxation, Resource, ResourceAllocation, ResourceRequest} +import io.gearpump.cluster.worker.WorkerId import io.gearpump.util.ActorSystemBooter.RegisterActorSystem import io.gearpump.util.ActorUtil -import org.scalatest.{BeforeAndAfter, Matchers, WordSpec} -class DistShellAppMasterSpec extends WordSpec with Matchers with BeforeAndAfter{ +class DistShellAppMasterSpec extends WordSpec with Matchers with BeforeAndAfter { implicit val system = ActorSystem("AppMasterSpec", TestUtil.DEFAULT_CONFIG) val mockMaster = TestProbe()(system) val mockWorker1 = TestProbe()(system) @@ -46,27 +50,29 @@ class DistShellAppMasterSpec extends WordSpec with Matchers with BeforeAndAfter{ "DistributedShell AppMaster" should { "launch one ShellTask on each worker" in { val appMasterInfo = AppMasterRuntimeInfo(appId, appName = appId.toString) - val appMasterContext = AppMasterContext(appId, userName, resource, null, appJar, masterProxy, appMasterInfo) + val appMasterContext = AppMasterContext(appId, userName, resource, null, appJar, + masterProxy, appMasterInfo) TestActorRef[DistShellAppMaster]( AppMasterRuntimeEnvironment.props(List(masterProxy.path), appDescription, appMasterContext)) mockMaster.expectMsgType[RegisterAppMaster] mockMaster.reply(AppMasterRegistered(appId)) - //The DistributedShell AppMaster will ask for worker list + // The DistributedShell AppMaster asks for worker list from Master. mockMaster.expectMsg(GetAllWorkers) mockMaster.reply(WorkerList(workerList)) - //After worker list is ready, DistributedShell AppMaster will request resouce on each worker + // After worker list is ready, DistributedShell AppMaster requests resource on each worker workerList.foreach { workerId => - mockMaster.expectMsg(RequestResource(appId, ResourceRequest(Resource(1), workerId, relaxation = Relaxation.SPECIFICWORKER))) + mockMaster.expectMsg(RequestResource(appId, ResourceRequest(Resource(1), workerId, + relaxation = Relaxation.SPECIFICWORKER))) } - mockMaster.reply(ResourceAllocated(Array(ResourceAllocation(resource, mockWorker1.ref, WorkerId(1, 0L))))) + mockMaster.reply(ResourceAllocated( + Array(ResourceAllocation(resource, mockWorker1.ref, WorkerId(1, 0L))))) mockWorker1.expectMsgClass(classOf[LaunchExecutor]) mockWorker1.reply(RegisterActorSystem(ActorUtil.getSystemAddress(system).toString)) } } after { - system.shutdown() - system.awaitTermination() + system.terminate() + Await.result(system.whenTerminated, Duration.Inf) } - } http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/DistributedShellClientSpec.scala ---------------------------------------------------------------------- diff --git a/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/DistributedShellClientSpec.scala b/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/DistributedShellClientSpec.scala index adefa40..973b3b3 100644 --- a/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/DistributedShellClientSpec.scala +++ b/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/DistributedShellClientSpec.scala @@ -7,7 +7,7 @@ * "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 + * 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, @@ -17,19 +17,20 @@ */ package io.gearpump.examples.distributedshell +import scala.concurrent.Future +import scala.util.{Success, Try} + import akka.testkit.TestProbe -import io.gearpump.cluster.ClientToMaster.ResolveAppId -import io.gearpump.cluster.MasterToClient.ResolveAppIdResult -import io.gearpump.cluster.{TestUtil, MasterHarness} -import DistShellAppMaster.ShellCommand -import io.gearpump.util.{LogUtil, Constants, Util} import org.scalatest.{BeforeAndAfter, Matchers, PropSpec} -import scala.concurrent.Future - -import scala.util.{Try, Success} +import io.gearpump.cluster.ClientToMaster.ResolveAppId +import io.gearpump.cluster.MasterToClient.ResolveAppIdResult +import io.gearpump.cluster.{MasterHarness, TestUtil} +import io.gearpump.examples.distributedshell.DistShellAppMaster.ShellCommand +import io.gearpump.util.LogUtil -class DistributedShellClientSpec extends PropSpec with Matchers with BeforeAndAfter with MasterHarness { +class DistributedShellClientSpec + extends PropSpec with Matchers with BeforeAndAfter with MasterHarness { private val LOG = LogUtil.getLogger(getClass) @@ -41,17 +42,19 @@ class DistributedShellClientSpec extends PropSpec with Matchers with BeforeAndAf shutdownActorSystem() } - override def config = TestUtil.DEFAULT_CONFIG + protected override def config = TestUtil.DEFAULT_CONFIG property("DistributedShellClient should succeed to submit application with required arguments") { val command = "ls /" val requiredArgs = Array("-appid", "0", "-command", command) val masterReceiver = createMockMaster() - assert(Try(DistributedShellClient.main(Array.empty[String])).isFailure, "missing required arguments, print usage") - + assert(Try(DistributedShellClient.main(Array.empty[String])).isFailure, + "missing required arguments, print usage") - Future {DistributedShellClient.main(masterConfig, requiredArgs)} + Future { + DistributedShellClient.main(masterConfig, requiredArgs) + } masterReceiver.expectMsg(PROCESS_BOOT_TIME, ResolveAppId(0)) val mockAppMaster = TestProbe()(getActorSystem) http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/DistributedShellSpec.scala ---------------------------------------------------------------------- diff --git a/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/DistributedShellSpec.scala b/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/DistributedShellSpec.scala index 2a369c4..6eeba58 100644 --- a/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/DistributedShellSpec.scala +++ b/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/DistributedShellSpec.scala @@ -7,7 +7,7 @@ * "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 + * 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, @@ -17,19 +17,19 @@ */ package io.gearpump.examples.distributedshell +import scala.concurrent.Future +import scala.util.Success + import com.typesafe.config.Config -import io.gearpump.cluster.ClientToMaster.SubmitApplication -import io.gearpump.cluster.MasterToClient.SubmitApplicationResult -import io.gearpump.cluster.{TestUtil, MasterHarness} -import io.gearpump.util.Util -import org.scalatest.{BeforeAndAfter, Matchers, PropSpec} import org.scalatest.prop.PropertyChecks +import org.scalatest.{BeforeAndAfter, Matchers, PropSpec} -import scala.util.{Try, Success} - -import scala.concurrent.Future +import io.gearpump.cluster.ClientToMaster.SubmitApplication +import io.gearpump.cluster.MasterToClient.SubmitApplicationResult +import io.gearpump.cluster.{MasterHarness, TestUtil} -class DistributedShellSpec extends PropSpec with PropertyChecks with Matchers with BeforeAndAfter with MasterHarness { +class DistributedShellSpec + extends PropSpec with PropertyChecks with Matchers with BeforeAndAfter with MasterHarness { before { startActorSystem() @@ -46,7 +46,9 @@ class DistributedShellSpec extends PropSpec with PropertyChecks with Matchers wi val masterReceiver = createMockMaster() - Future{DistributedShell.main(masterConfig, requiredArgs)} + Future { + DistributedShell.main(masterConfig, requiredArgs) + } masterReceiver.expectMsgType[SubmitApplication](PROCESS_BOOT_TIME) masterReceiver.reply(SubmitApplicationResult(Success(0))) http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/ShellCommandResultAggregatorSpec.scala ---------------------------------------------------------------------- diff --git a/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/ShellCommandResultAggregatorSpec.scala b/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/ShellCommandResultAggregatorSpec.scala index 6bb1105..d59981b 100644 --- a/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/ShellCommandResultAggregatorSpec.scala +++ b/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/ShellCommandResultAggregatorSpec.scala @@ -7,7 +7,7 @@ * "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 + * 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, @@ -18,8 +18,8 @@ package io.gearpump.examples.distributedshell import org.scalatest.{BeforeAndAfter, Matchers, WordSpec} -import DistShellAppMaster.{ShellCommandResultAggregator, ShellCommandResult} +import io.gearpump.examples.distributedshell.DistShellAppMaster.{ShellCommandResult, ShellCommandResultAggregator} class ShellCommandResultAggregatorSpec extends WordSpec with Matchers with BeforeAndAfter { "ShellCommandResultAggregator" should { http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/ShellExecutorSpec.scala ---------------------------------------------------------------------- diff --git a/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/ShellExecutorSpec.scala b/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/ShellExecutorSpec.scala index d51880b..b301973 100644 --- a/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/ShellExecutorSpec.scala +++ b/examples/distributedshell/src/test/scala/io/gearpump/examples/distributedshell/ShellExecutorSpec.scala @@ -7,7 +7,7 @@ * "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 + * 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, @@ -17,18 +17,20 @@ */ package io.gearpump.examples.distributedshell +import scala.concurrent.Await +import scala.concurrent.duration.Duration +import scala.sys.process._ +import scala.util.{Failure, Success, Try} + import akka.actor.{ActorSystem, Props} import akka.testkit.TestProbe -import io.gearpump.WorkerId -import io.gearpump.examples.distributedshell.DistShellAppMaster.ShellCommandResult +import org.scalatest.{Matchers, WordSpec} + import io.gearpump.cluster.appmaster.WorkerInfo import io.gearpump.cluster.scheduler.Resource +import io.gearpump.cluster.worker.WorkerId import io.gearpump.cluster.{ExecutorContext, TestUtil, UserConfig} -import DistShellAppMaster.{ShellCommand, ShellCommandResult} -import org.scalatest.{Matchers, WordSpec} - -import scala.sys.process._ -import scala.util.{Failure, Success, Try} +import io.gearpump.examples.distributedshell.DistShellAppMaster.{ShellCommand, ShellCommandResult} class ShellExecutorSpec extends WordSpec with Matchers { @@ -43,10 +45,12 @@ class ShellExecutorSpec extends WordSpec with Matchers { val mockMaster = TestProbe()(system) val worker = TestProbe() val workerInfo = WorkerInfo(workerId, worker.ref) - val executorContext = ExecutorContext(executorId, workerInfo, appId, appName, mockMaster.ref, resource) - val executor = system.actorOf(Props(classOf[ShellExecutor], executorContext, UserConfig.empty)) + val executorContext = ExecutorContext(executorId, workerInfo, appId, appName, + mockMaster.ref, resource) + val executor = system.actorOf(Props(classOf[ShellExecutor], executorContext, + UserConfig.empty)) - val process = Try(s"ls /" !!) + val process = Try(s"ls /".!!) val result = process match { case Success(msg) => msg case Failure(ex) => ex.getMessage @@ -55,8 +59,8 @@ class ShellExecutorSpec extends WordSpec with Matchers { assert(mockMaster.receiveN(1).head.asInstanceOf[ShellCommandResult].equals( ShellCommandResult(executorId, result))) - system.shutdown() - system.awaitTermination() + system.terminate() + Await.result(system.whenTerminated, Duration.Inf) } } } http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/examples/distributeservice/README.md ---------------------------------------------------------------------- diff --git a/examples/distributeservice/README.md b/examples/distributeservice/README.md index da5acac..82b3726 100644 --- a/examples/distributeservice/README.md +++ b/examples/distributeservice/README.md @@ -15,7 +15,7 @@ In order to run the example: -script ${Script_Path} -serviceName ${Service_Name} -target ${Target_Path} -Dkey1=value1 -Dkey2=value2 ```<br> This command will distribute the service zip file(variable ```file```) to the target path(variable ```target```), then copy the script to - ```/etc/init.d``` on each machine and install this servcie named with ```serviceName```<br> + ```/etc/init.d``` on each machine and install this service named with ```serviceName```<br> Note that you can pass some variables when the script file is installed, for example, you can submit a script template with syntax like ``` role=${${hostname}.role} http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/DistServiceAppMaster.scala ---------------------------------------------------------------------- diff --git a/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/DistServiceAppMaster.scala b/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/DistServiceAppMaster.scala index 4d3492a..a220dc6 100644 --- a/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/DistServiceAppMaster.scala +++ b/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/DistServiceAppMaster.scala @@ -7,7 +7,7 @@ * "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 + * 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, @@ -18,21 +18,22 @@ package io.gearpump.experiments.distributeservice import java.io.File +import scala.concurrent.Future import akka.actor.{Deploy, Props} import akka.pattern.{ask, pipe} import akka.remote.RemoteScope import com.typesafe.config.Config +import org.slf4j.Logger + import io.gearpump.cluster.ClientToMaster.ShutdownApplication import io.gearpump.cluster.appmaster.ExecutorSystemScheduler.{ExecutorSystemJvmConfig, ExecutorSystemStarted, StartExecutorSystemTimeout} import io.gearpump.cluster.{AppDescription, AppMasterContext, ApplicationMaster, ExecutorContext} -import DistServiceAppMaster.{FileContainer, GetFileContainer, InstallService} +import io.gearpump.experiments.distributeservice.DistServiceAppMaster.{FileContainer, GetFileContainer, InstallService} import io.gearpump.util._ -import org.slf4j.Logger - -import scala.concurrent.Future -class DistServiceAppMaster(appContext : AppMasterContext, app : AppDescription) extends ApplicationMaster { +class DistServiceAppMaster(appContext: AppMasterContext, app: AppDescription) + extends ApplicationMaster { import appContext._ import context.dispatcher implicit val timeout = Constants.FUTURE_TIMEOUT @@ -42,7 +43,7 @@ class DistServiceAppMaster(appContext : AppMasterContext, app : AppDescription) val rootDirectory = new File("/") val host = context.system.settings.config.getString(Constants.GEARPUMP_HOSTNAME) - val server = context.actorOf(Props(classOf[FileServer], rootDirectory, host , 0)) + val server = context.actorOf(Props(classOf[FileServer], rootDirectory, host, 0)) override def preStart(): Unit = { LOG.info(s"Distribute Service AppMaster started") @@ -54,10 +55,12 @@ class DistServiceAppMaster(appContext : AppMasterContext, app : AppDescription) override def receive: Receive = { case ExecutorSystemStarted(executorSystem, _) => import executorSystem.{address, resource => executorResource, worker} - val executorContext = ExecutorContext(currentExecutorId, worker, appId, app.name, self, executorResource) - //start executor - val executor = context.actorOf(Props(classOf[DistServiceExecutor], executorContext, app.userConfig) - .withDeploy(Deploy(scope = RemoteScope(address))), currentExecutorId.toString) + val executorContext = ExecutorContext(currentExecutorId, worker, + appId, app.name, self, executorResource) + // start executor + val executor = context.actorOf(Props(classOf[DistServiceExecutor], + executorContext, app.userConfig).withDeploy( + Deploy(scope = RemoteScope(address))), currentExecutorId.toString) executorSystem.bindLifeCycleWith(executor) currentExecutorId += 1 case StartExecutorSystemTimeout => @@ -75,7 +78,8 @@ class DistServiceAppMaster(appContext : AppMasterContext, app : AppDescription) private def getExecutorJvmConfig: ExecutorSystemJvmConfig = { val config: Config = app.clusterConfig - val jvmSetting = Util.resolveJvmSetting(config.withFallback(context.system.settings.config)).executor + val jvmSetting = Util.resolveJvmSetting( + config.withFallback(context.system.settings.config)).executor ExecutorSystemJvmConfig(jvmSetting.classPath, jvmSetting.vmargs, appJar, username, config) } @@ -87,10 +91,10 @@ object DistServiceAppMaster { case class FileContainer(url: String) case class InstallService( - url: String, - zipFileName: String, - targetPath: String, - script : Array[Byte], - serviceName: String, - serviceSettings: Map[String, Any]) + url: String, + zipFileName: String, + targetPath: String, + script: Array[Byte], + serviceName: String, + serviceSettings: Map[String, Any]) } http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/DistServiceExecutor.scala ---------------------------------------------------------------------- diff --git a/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/DistServiceExecutor.scala b/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/DistServiceExecutor.scala index 7fc2a94..4a2a876 100644 --- a/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/DistServiceExecutor.scala +++ b/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/DistServiceExecutor.scala @@ -7,7 +7,7 @@ * "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 + * 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, @@ -17,29 +17,30 @@ */ package io.gearpump.experiments.distributeservice -import java.io.{FileWriter, File} +import java.io.{File, FileWriter} import java.net.InetAddress +import scala.collection.JavaConverters._ +import scala.io.Source +import scala.sys.process._ +import scala.util.{Failure, Success, Try} import akka.actor.Actor import org.apache.commons.io.FileUtils import org.apache.commons.lang.text.StrSubstitutor -import io.gearpump.cluster.{UserConfig, ExecutorContext} -import DistServiceAppMaster.InstallService -import io.gearpump.util.{ActorUtil, LogUtil} import org.slf4j.Logger -import scala.io.Source -import scala.sys.process._ -import collection.JavaConversions._ -import scala.util.{Failure, Success, Try} +import io.gearpump.cluster.{ExecutorContext, UserConfig} +import io.gearpump.experiments.distributeservice.DistServiceAppMaster.InstallService +import io.gearpump.util.{ActorUtil, LogUtil} -class DistServiceExecutor(executorContext: ExecutorContext, userConf : UserConfig) extends Actor { +class DistServiceExecutor(executorContext: ExecutorContext, userConf: UserConfig) extends Actor { import executorContext._ private val LOG: Logger = LogUtil.getLogger(getClass, executor = executorId, app = appId) override def receive: Receive = { case InstallService(url, zipFileName, targetPath, scriptData, serviceName, serviceSettings) => - LOG.info(s"Executor $executorId receive command to install service $serviceName to $targetPath") + LOG.info(s"Executor $executorId receive command to install " + + s"service $serviceName to $targetPath") unzipFile(url, zipFileName, targetPath) installService(scriptData, serviceName, serviceSettings) } @@ -47,31 +48,32 @@ class DistServiceExecutor(executorContext: ExecutorContext, userConf : UserConfi private def unzipFile(url: String, zipFileName: String, targetPath: String) = { val zipFile = File.createTempFile(System.currentTimeMillis().toString, zipFileName) val dir = new File(targetPath) - if(dir.exists()) { + if (dir.exists()) { FileUtils.forceDelete(dir) } val bytes = FileServer.newClient.get(url).get FileUtils.writeByteArrayToFile(zipFile, bytes) - val result = Try(s"unzip ${zipFile.getAbsolutePath} -d $targetPath" !!) + val result = Try(s"unzip ${zipFile.getAbsolutePath} -d $targetPath".!!) result match { case Success(msg) => LOG.info(s"Executor $executorId unzip file to $targetPath") - case Failure(ex) => throw ex + case Failure(ex) => throw ex } } - private def installService(scriptData: Array[Byte], serviceName: String, serviceSettings: Map[String, Any]) = { + private def installService( + scriptData: Array[Byte], serviceName: String, serviceSettings: Map[String, Any]) = { val tempFile = File.createTempFile("gearpump", serviceName) FileUtils.writeByteArrayToFile(tempFile, scriptData) val script = new File("/etc/init.d", serviceName) writeFileWithEnvVariables(tempFile, script, serviceSettings ++ getEnvSettings) - val result = Try(s"chkconfig --add $serviceName" !!) + val result = Try(s"chkconfig --add $serviceName".!!) result match { case Success(msg) => LOG.info(s"Executor install service $serviceName successfully!") case Failure(ex) => throw ex } } - private def getEnvSettings : Map[String, Any] = { + private def getEnvSettings: Map[String, Any] = { Map("workerId" -> worker, "localhost" -> ActorUtil.getSystemAddress(context.system).host.get, "hostname" -> InetAddress.getLocalHost.getHostName) @@ -79,7 +81,7 @@ class DistServiceExecutor(executorContext: ExecutorContext, userConf : UserConfi private def writeFileWithEnvVariables(source: File, target: File, envs: Map[String, Any]) = { val writer = new FileWriter(target) - val sub = new StrSubstitutor(mapAsJavaMap(envs)) + val sub = new StrSubstitutor(envs.asJava) sub.setEnableSubstitutionInVariables(true) Source.fromFile(source).getLines().foreach(line => writer.write(sub.replace(line) + "\r\n")) writer.close() http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/DistributeService.scala ---------------------------------------------------------------------- diff --git a/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/DistributeService.scala b/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/DistributeService.scala index d558b5d..522dc5e 100644 --- a/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/DistributeService.scala +++ b/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/DistributeService.scala @@ -7,7 +7,7 @@ * "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 + * 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, @@ -17,13 +17,15 @@ */ package io.gearpump.experiments.distributeservice +import org.slf4j.Logger + import io.gearpump.cluster.client.ClientContext -import io.gearpump.cluster.{Application, AppJar, UserConfig, AppDescription} -import io.gearpump.cluster.main.{ParseResult, CLIOption, ArgumentsParser} +import io.gearpump.cluster.main.{ArgumentsParser, CLIOption} +import io.gearpump.cluster.{Application, UserConfig} import io.gearpump.util.{AkkaApp, LogUtil} -import org.slf4j.Logger -object DistributeService extends AkkaApp with ArgumentsParser { +/** Demo app to remotely deploy and start system service on machines in the cluster */ +object DistributeService extends AkkaApp with ArgumentsParser { private val LOG: Logger = LogUtil.getLogger(getClass) override val options: Array[(String, CLIOption[Any])] = Array.empty @@ -31,7 +33,8 @@ object DistributeService extends AkkaApp with ArgumentsParser { override def main(akkaConf: Config, args: Array[String]): Unit = { LOG.info(s"Distribute Service submitting application...") val context = ClientContext(akkaConf) - val appId = context.submit(Application[DistServiceAppMaster]("DistributedService", UserConfig.empty)) + val appId = context.submit(Application[DistServiceAppMaster]("DistributedService", + UserConfig.empty)) context.close() LOG.info(s"Distribute Service Application started with appId $appId !") } http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/DistributeServiceClient.scala ---------------------------------------------------------------------- diff --git a/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/DistributeServiceClient.scala b/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/DistributeServiceClient.scala index 5f4ebbb..0d85001 100644 --- a/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/DistributeServiceClient.scala +++ b/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/DistributeServiceClient.scala @@ -7,7 +7,7 @@ * "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 + * 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, @@ -18,31 +18,35 @@ package io.gearpump.experiments.distributeservice import java.io.File -import org.apache.commons.io.FileUtils -import io.gearpump.cluster.client.ClientContext -import io.gearpump.cluster.main.{CLIOption, ArgumentsParser} -import DistServiceAppMaster.{InstallService, FileContainer, GetFileContainer} -import io.gearpump.util.{AkkaApp, LogUtil, FileServer, Constants} -import org.slf4j.{LoggerFactory, Logger} - -import akka.pattern.ask import scala.concurrent.Future import scala.util.{Failure, Success} -object DistributeServiceClient extends AkkaApp with ArgumentsParser{ +import akka.pattern.ask +import org.apache.commons.io.FileUtils + +import io.gearpump.cluster.client.ClientContext +import io.gearpump.cluster.main.{ArgumentsParser, CLIOption} +import io.gearpump.experiments.distributeservice.DistServiceAppMaster.{FileContainer, GetFileContainer, InstallService} +import io.gearpump.util.{AkkaApp, Constants} + +/** Client to submit the service jar */ +object DistributeServiceClient extends AkkaApp with ArgumentsParser { implicit val timeout = Constants.FUTURE_TIMEOUT override val options: Array[(String, CLIOption[Any])] = Array( "appid" -> CLIOption[Int]("<the distributed shell appid>", required = true), "file" -> CLIOption[String]("<service zip file path>", required = true), - "script" -> CLIOption[String]("<file path of service script that will be installed to /etc/init.d>", required = true), + "script" -> CLIOption[String]( + "<file path of service script that will be installed to /etc/init.d>", required = true), "serviceName" -> CLIOption[String]("<service name>", required = true), "target" -> CLIOption[String]("<target path on each machine>", required = true) ) - override def help : Unit = { - super.help + override def help(): Unit = { + super.help() + // scalastyle:off println Console.err.println(s"-D<name>=<value> set a property to the service") + // scalastyle:on println } override def main(akkaConf: Config, args: Array[String]): Unit = { @@ -75,7 +79,7 @@ object DistributeServiceClient extends AkkaApp with ArgumentsParser{ private def parseServiceConfig(args: Array[String]): Map[String, Any] = { val result = Map.empty[String, Any] args.foldLeft(result) { (result, argument) => - if(argument.startsWith("-D") && argument.contains("=")) { + if (argument.startsWith("-D") && argument.contains("=")) { val fixedKV = argument.substring(2).split("=") result + (fixedKV(0) -> fixedKV(1)) } else { http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/c176e448/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/FileServer.scala ---------------------------------------------------------------------- diff --git a/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/FileServer.scala b/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/FileServer.scala index 893aa19..ed0b24d 100644 --- a/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/FileServer.scala +++ b/examples/distributeservice/src/main/scala/io/gearpump/experiments/distributeservice/FileServer.scala @@ -7,7 +7,7 @@ * "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 + * 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, @@ -16,21 +16,20 @@ * limitations under the License. */ - package io.gearpump.experiments.distributeservice import java.io.File +import scala.util.{Failure, Success, Try} import akka.actor.{Actor, Stash} import akka.io.IO -import io.gearpump.util.LogUtil import org.apache.commons.httpclient.HttpClient import org.apache.commons.httpclient.methods.{ByteArrayRequestEntity, GetMethod, PostMethod} import spray.can.Http import spray.http.HttpMethods._ import spray.http._ -import io.gearpump.util.FileUtils -import scala.util.{Failure, Success, Try} + +import io.gearpump.util.{FileUtils, LogUtil} /** * @@ -38,18 +37,18 @@ import scala.util.{Failure, Success, Try} * * port: set port to 0 if you want to bind to random port */ -class FileServer(rootDir: File, host: String, port : Int) extends Actor with Stash { +class FileServer(rootDir: File, host: String, port: Int) extends Actor with Stash { private val LOG = LogUtil.getLogger(getClass) implicit val system = context.system override def preStart(): Unit = { - // create http server + // Creates http server IO(Http) ! Http.Bind(self, host, port) } - override def postStop() : Unit = { - //stop the server + override def postStop(): Unit = { + // Stop the server IO(Http) ! Http.Unbind } @@ -62,14 +61,14 @@ class FileServer(rootDir: File, host: String, port : Int) extends Actor with Sta stash() } - def listen(port : Int) : Receive = { + def listen(port: Int): Receive = { case FileServer.GetPort => { sender ! FileServer.Port(port) } case Http.Connected(remote, _) => sender ! Http.Register(self) - // fetch file from remote uri + // Fetches files from remote uri case HttpRequest(GET, uri, _, _, _) => val child = uri.path.toString() val payload = Try { @@ -83,8 +82,8 @@ class FileServer(rootDir: File, host: String, port : Int) extends Actor with Sta LOG.error("failed to get file " + ex.getMessage) sender ! HttpResponse(status = StatusCodes.InternalServerError, entity = ex.getMessage) } - //save file to remote uri - case post @ HttpRequest(POST, uri, _, _, _) => + // Save file to remote uri + case post@HttpRequest(POST, uri, _, _, _) => val child = uri.path.toString() val status = Try { @@ -104,14 +103,14 @@ class FileServer(rootDir: File, host: String, port : Int) extends Actor with Sta object FileServer { object GetPort - case class Port(port : Int) + case class Port(port: Int) - def newClient = new Client + def newClient: Client = new Client class Client { val client = new HttpClient() - def save(uri : String, data : Array[Byte]) : Try[Int] = { + def save(uri: String, data: Array[Byte]): Try[Int] = { Try { val post = new PostMethod(uri) val entity = new ByteArrayRequestEntity(data) @@ -120,7 +119,7 @@ object FileServer { } } - def get(uri : String) : Try[Array[Byte]] = { + def get(uri: String): Try[Array[Byte]] = { val get = new GetMethod(uri) val status = Try { client.executeMethod(get)
