http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/0f11a591/eagle-core/eagle-app/eagle-stream-application-manager/src/test/resources/application.conf ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-app/eagle-stream-application-manager/src/test/resources/application.conf b/eagle-core/eagle-app/eagle-stream-application-manager/src/test/resources/application.conf deleted file mode 100644 index 4c21a7c..0000000 --- a/eagle-core/eagle-app/eagle-stream-application-manager/src/test/resources/application.conf +++ /dev/null @@ -1,42 +0,0 @@ -# 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. - - -### scheduler propertise -appCommandLoaderIntervalSecs = 1 -appHealthCheckIntervalSecs = 5 - -### execution platform properties -envContextConfig.env = "storm" -envContextConfig.url = "http://sandbox.hortonworks.com:8744"; -envContextConfig.nimbusHost = "sandbox.hortonworks.com" -envContextConfig.nimbusThriftPort = 6627 -envContextConfig.jarFile = "/dir-to-jar/eagle-topology-0.3.0-incubating-assembly.jar" - -### default topology properties -eagleProps.mailHost = "mailHost.com" -eagleProps.mailSmtpPort = "25" -eagleProps.mailDebug = "true" -eagleProps.eagleService.host = "localhost" -eagleProps.eagleService.port = 9099 -eagleProps.eagleService.username = "admin" -eagleProps.eagleService.password = "secret" -eagleProps.dataJoinPollIntervalSec = 30 - -dynamicConfigSource.enabled = true -dynamicConfigSource.initDelayMillis = 0 -dynamicConfigSource.delayMillis = 30000 - -
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/0f11a591/eagle-core/eagle-app/eagle-stream-application-manager/src/test/resources/log4j.properties ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-app/eagle-stream-application-manager/src/test/resources/log4j.properties b/eagle-core/eagle-app/eagle-stream-application-manager/src/test/resources/log4j.properties deleted file mode 100644 index 25331ab..0000000 --- a/eagle-core/eagle-app/eagle-stream-application-manager/src/test/resources/log4j.properties +++ /dev/null @@ -1,35 +0,0 @@ -# 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. - -log4j.rootLogger=INFO, stdout - - eagle.log.dir=../logs - eagle.log.file=eagle.log - -# standard output -log4j.appender.stdout=org.apache.log4j.ConsoleAppender -log4j.appender.stdout.layout=org.apache.log4j.PatternLayout -log4j.appender.stdout.layout.ConversionPattern=%d{ISO8601} %p [%t] %c{2}[%L]: %m%n - -# Daily Rolling File Appender - log4j.appender.DRFA=org.apache.log4j.DailyRollingFileAppender - log4j.appender.DRFA.File=${eagle.log.dir}/${eagle.log.file} - log4j.appender.DRFA.DatePattern=.yyyy-MM-dd -## 30-day backup -# log4j.appender.DRFA.MaxBackupIndex=30 - log4j.appender.DRFA.layout=org.apache.log4j.PatternLayout - -# Pattern format: Date LogLevel LoggerName LogMessage -log4j.appender.DRFA.layout.ConversionPattern=%d{ISO8601} %p [%t] %c{2}[%L]: %m%n \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/0f11a591/eagle-core/eagle-app/eagle-stream-application-manager/src/test/scala/org/apache/eagle/stream/application/scheduler/MockTopology.scala ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-app/eagle-stream-application-manager/src/test/scala/org/apache/eagle/stream/application/scheduler/MockTopology.scala b/eagle-core/eagle-app/eagle-stream-application-manager/src/test/scala/org/apache/eagle/stream/application/scheduler/MockTopology.scala deleted file mode 100644 index e87ee92..0000000 --- a/eagle-core/eagle-app/eagle-stream-application-manager/src/test/scala/org/apache/eagle/stream/application/scheduler/MockTopology.scala +++ /dev/null @@ -1,30 +0,0 @@ -/* - * 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. - * - */ - -package org.apache.eagle.stream.application.scheduler - -import com.typesafe.config.Config -import org.apache.eagle.stream.application.TopologyExecutable -import org.slf4j.LoggerFactory - -class MockTopology extends TopologyExecutable { - private val LOG = LoggerFactory.getLogger(classOf[MockTopology]) - override def submit(topology: String, config: Config): Unit = { - LOG.info(s"$topology is running") - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/0f11a591/eagle-core/eagle-app/eagle-stream-application-manager/src/test/scala/org/apache/eagle/stream/application/scheduler/StormApplicationManagerSpec.scala ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-app/eagle-stream-application-manager/src/test/scala/org/apache/eagle/stream/application/scheduler/StormApplicationManagerSpec.scala b/eagle-core/eagle-app/eagle-stream-application-manager/src/test/scala/org/apache/eagle/stream/application/scheduler/StormApplicationManagerSpec.scala deleted file mode 100644 index 1cad3a7..0000000 --- a/eagle-core/eagle-app/eagle-stream-application-manager/src/test/scala/org/apache/eagle/stream/application/scheduler/StormApplicationManagerSpec.scala +++ /dev/null @@ -1,40 +0,0 @@ -package org.apache.eagle.stream.application.scheduler - -import com.typesafe.config.ConfigFactory -import org.apache.eagle.common.config.EagleConfigConstants -import org.apache.eagle.stream.application.ExecutionPlatform -import org.apache.eagle.stream.application.impl.StormExecutionPlatform - -/* - * 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. - * - */ - - -object StormApplicationManagerSpec extends App { - val manager: ExecutionPlatform = new StormExecutionPlatform - val baseConfig = ConfigFactory.load() - val topoConfigStr: String = "webConfig{\"hbase.zookeeper.property.clientPort\":\"2181\", \"hbase.zookeeper.quorum\":\"localhost\"}\nappConfig{\n \"envContextConfig\" : {\n \"env\" : \"storm\",\n \"mode\" : \"cluster\",\n \"topologyName\" : \"sandbox-hbaseSecurityLog-topology\",\n \"stormConfigFile\" : \"security-auditlog-storm.yaml\",\n \"parallelismConfig\" : {\n \"kafkaMsgConsumer\" : 1,\n \"hbaseSecurityLogAlertExecutor*\" : 1\n }\n },\n \"dataSourceConfig\": {\n \"topic\" : \"sandbox_hbase_security_log\",\n \"zkConnection\" : \"127.0.0.1:2181\",\n \"zkConnectionTimeoutMS\" : 15000,\n \"brokerZkPath\" : \"/brokers\",\n \"fetchSize\" : 1048586,\n \"deserializerClass\" : \"org.apache.eagle.security.hbase.parse.HbaseAuditLogKafkaDeserializer\",\n \"transactionZKServers\" : \"127.0.0.1\",\n \"transactionZKPort\" : 2181,\n \"transactionZKRoot\" : \"/consumers\",\n \"consumerGroupId\" : \"eagle.hbasesecurity.consumer\",\n \"transactionStateUpdateMS\" : 2000\n },\n \"alertExecutorConfigs\" : {\n \"hbaseSecurityLogAlertExecutor\" : {\n \"parallelism\" : 1,\n \"partitioner\" : \"org.apache.eagle.policy.DefaultPolicyPartitioner\"\n \"needValidation\" : \"true\"\n }\n },\n \"eagleProps\" : {\n \"site\" : \"sandbox\",\n \"application\": \"hbaseSecurityLog\",\n \"dataJoinPollIntervalSec\" : 30,\n \"mailHost\" : \"mailHost.com\",\n \"mailSmtpPort\":\"25\",\n \"mailDebug\" : \"true\",\n \"eagleService\": {\n \"host\": \"localhost\",\n \"port\": 9099\n \"username\": \"admin\",\n \"password\": \"secret\"\n }\n },\n \"dynamicConfigSource\" : {\n \"enabled\" : true,\n \"initDelayMillis\" : 0,\n \"delayMillis\" : 30000\n }\n}" - - val topoConfig = ConfigFactory.parseString(topoConfigStr) - val conf = topoConfig.getConfig(EagleConfigConstants.APP_CONFIG).withFallback(baseConfig) - - //val (ret, nextState) = manager.execute("START", topologyDescModel, null, conf) - //println(s"Result: ret=$ret, nextState=$nextState") -} - - - http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/0f11a591/eagle-core/eagle-app/eagle-stream-application-manager/src/test/scala/org/apache/eagle/stream/application/scheduler/TestScheduler.scala ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-app/eagle-stream-application-manager/src/test/scala/org/apache/eagle/stream/application/scheduler/TestScheduler.scala b/eagle-core/eagle-app/eagle-stream-application-manager/src/test/scala/org/apache/eagle/stream/application/scheduler/TestScheduler.scala deleted file mode 100644 index 3db2d67..0000000 --- a/eagle-core/eagle-app/eagle-stream-application-manager/src/test/scala/org/apache/eagle/stream/application/scheduler/TestScheduler.scala +++ /dev/null @@ -1,61 +0,0 @@ -/** - * 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. - */ - -package org.apache.eagle.stream.application.scheduler - -import akka.actor.{ActorSystem, Props} -import akka.testkit.{TestActorRef, TestKit} -import com.typesafe.config.ConfigFactory -import org.scalatest.{Ignore, BeforeAndAfterAll, MustMatchers, WordSpecLike} - -@Ignore -class TestSchedulerSpec extends TestKit(ActorSystem("stream-app-scheduler")) -with WordSpecLike with MustMatchers with BeforeAndAfterAll { - - "A Scheduler actor" must { - "Forward a message it receives" in { - val coordinator = TestActorRef[StreamAppCoordinator] - coordinator ! CommandLoaderEvent - expectNoMsg() - } - } - - "A Integrated test" must { - "run end-to-end" in { - val coordinator = system.actorOf(Props[StreamAppCoordinator]) - coordinator ! CommandLoaderEvent - expectNoMsg() - } - } - - override def afterAll(): Unit = { - super.afterAll() - system.shutdown() - } -} - -@Ignore -object TestStreamAppScheduler extends App { - val conf: String = """ - akka.loglevel = "DEBUG" - akka.actor.debug { - receive = on - lifecycle = on - } - """ - new ApplicationScheduler().start(ConfigFactory.parseString(conf)) -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/0f11a591/eagle-core/eagle-app/pom.xml ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-app/pom.xml b/eagle-core/eagle-app/pom.xml index 6f3069c..5637b01 100644 --- a/eagle-core/eagle-app/pom.xml +++ b/eagle-core/eagle-app/pom.xml @@ -32,7 +32,6 @@ <modules> <module>eagle-app-base</module> - <module>eagle-stream-application-manager</module> <module>eagle-application-service</module> </modules> </project> http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/0f11a591/eagle-core/eagle-data-process/eagle-stream-pipeline/README.md ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-data-process/eagle-stream-pipeline/README.md b/eagle-core/eagle-data-process/eagle-stream-pipeline/README.md deleted file mode 100644 index b8a0bdc..0000000 --- a/eagle-core/eagle-data-process/eagle-stream-pipeline/README.md +++ /dev/null @@ -1,80 +0,0 @@ -<!-- -{% comment %} -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. -{% endcomment %} ---> - -Eagle Declarative Streaming DSL -=============================== - -DSL Format ----------- - - { - config { - config.key = configValue - } - - schema { - metricStreamSchema { - metric: string - value: double - timestamp: long - } - } - - dataflow { - kafkaSource.source1 { - schema = "metricStreamSchema" - } - kafkaSource.source2 { - schema = { - metric: string - value: double - timestamp: long - } - } - } - } - -Usage ------ - - val pipeline = Pipeline.parseResource("pipeline.conf") - val stream = Pipeline.compile(pipeline) - stream.submit[storm] - -Features --------- -* [x] Compile DSL Configure to Pipeline model -* [x] Compile Pipeline model to Stream Execution Graph -* [x] Submit Stream Execution Graph to actual running environment say storm -* [x] Support Alert and Persistence for metric monitoring -* [ ] Extensible stream module management and automatically scan and register module -* [x] Pipeline runner CLI tool and shell script -* [ ] Decouple pipeline compiler and scheduler into individual modules -* [ ] Stream Pipeline Scheduler -* [ ] Graph editor to define streaming graph in UI -* [?] JSON/Config & Scala Case Class Mapping (https://github.com/scala/pickling) -* [?] Raw message structure oriented programing is a little ugly, we should define a generic message/event consist of [payload:stream/timestamp/serializer/deserializer,data:message] -* [ ] Provide stream schema inline and send to metadata when submitting -* [ ] UI should support specify executorId when defining new stream -* [ ] Lack of a entity named StreamEntity for the workflow of defining topology&policy end-to-end -* [!] Fix configuration conflict, should pass through Config instead of ConfigFactory.load() manually -* [ ] Override application configuration with pipeline configuration -* [ ] Refactor schema registration structure and automatically submit stream schema when submitting pipeline -* [ ] Submit alertStream, alertExecutorId mapping to AlertExecutorService when submitting pipeline -* [x] Supports `inputs` field to define connector \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/0f11a591/eagle-core/eagle-data-process/eagle-stream-pipeline/pom.xml ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-data-process/eagle-stream-pipeline/pom.xml b/eagle-core/eagle-data-process/eagle-stream-pipeline/pom.xml deleted file mode 100644 index 18fb610..0000000 --- a/eagle-core/eagle-data-process/eagle-stream-pipeline/pom.xml +++ /dev/null @@ -1,156 +0,0 @@ -<?xml version="1.0" encoding="UTF-8"?> -<!-- - ~ 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. - --> -<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> - <parent> - <artifactId>eagle-data-process-parent</artifactId> - <groupId>org.apache.eagle</groupId> - <version>0.5.0-incubating-SNAPSHOT</version> - </parent> - <modelVersion>4.0.0</modelVersion> - <artifactId>eagle-stream-pipeline</artifactId> - <dependencies> - <!--<dependency>--> - <!--<groupId>org.reflections</groupId>--> - <!--<artifactId>reflections</artifactId>--> - <!--</dependency>--> - <dependency> - <groupId>org.apache.eagle</groupId> - <artifactId>eagle-service-base</artifactId> - <version>${project.version}</version> - <exclusions> - <exclusion> - <groupId>org.ow2.asm</groupId> - <artifactId>asm-all</artifactId> - </exclusion> - <exclusion> - <groupId>asm</groupId> - <artifactId>asm</artifactId> - </exclusion> - <exclusion> - <groupId>asm</groupId> - <artifactId>asm-all</artifactId> - </exclusion> - <exclusion> - <groupId>asm</groupId> - <artifactId>asm-commons</artifactId> - </exclusion> - <exclusion> - <groupId>asm</groupId> - <artifactId>asm-tree</artifactId> - </exclusion> - <exclusion> - <artifactId>servlet-api</artifactId> - <groupId>javax.servlet</groupId> - </exclusion> - <exclusion> - <artifactId>slf4j-log4j12</artifactId> - <groupId>org.slf4j</groupId> - </exclusion> - </exclusions> - </dependency> - <dependency> - <groupId>org.apache.eagle</groupId> - <artifactId>eagle-storage-base</artifactId> - <version>${project.version}</version> - </dependency> - <dependency> - <groupId>log4j</groupId> - <artifactId>log4j</artifactId> - </dependency> -<!-- <dependency> - <groupId>org.slf4j</groupId> - <artifactId>slf4j-api</artifactId> - <scope>test</scope> - </dependency> ---> - <!--<dependency>--> - <!--<groupId>org.scala-lang</groupId>--> - <!--<artifactId>scala-reflect</artifactId>--> - <!--</dependency>--> - <!--<dependency>--> - <!--<groupId>org.scala-lang</groupId>--> - <!--<artifactId>scala-compiler</artifactId>--> - <!--<version>${scala.version}.0</version>--> - <!--</dependency>--> - <dependency> - <groupId>org.scalatest</groupId> - <artifactId>scalatest_${scala.version}</artifactId> - <scope>test</scope> - </dependency> - <dependency> - <groupId>org.apache.eagle</groupId> - <artifactId>eagle-stream-process-api</artifactId> - <version>${project.version}</version> - </dependency> - <!--<dependency>--> - <!--<groupId>com.typesafe.akka</groupId>--> - <!--<artifactId>akka-testkit_${scala.version}</artifactId>--> - <!--<version>${akka.actor.version}</version>--> - <!--<scope>test</scope>--> - <!--</dependency>--> - </dependencies> - <build> - <plugins> - <plugin> - <groupId>org.scala-tools</groupId> - <artifactId>maven-scala-plugin</artifactId> - <executions> - <execution> - <id>scala-compile-first</id> - <phase>process-resources</phase> - <goals> - <goal>add-source</goal> - <goal>compile</goal> - </goals> - </execution> - <execution> - <id>scala-test-compile</id> - <phase>process-test-resources</phase> - <goals> - <goal>testCompile</goal> - </goals> - </execution> - </executions> - </plugin> - <plugin> - <groupId>org.apache.maven.plugins</groupId> - <artifactId>maven-surefire-plugin</artifactId> - <configuration> - <skipTests>true</skipTests> - </configuration> - </plugin> - <plugin> - <groupId>org.scalatest</groupId> - <artifactId>scalatest-maven-plugin</artifactId> - <configuration> - <reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory> - <junitxml>.</junitxml> - <filereports>TestSuite.txt</filereports> - </configuration> - <executions> - <execution> - <id>test</id> - <goals> - <goal>test</goal> - </goals> - </execution> - </executions> - </plugin> - </plugins> - </build> -</project> http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/0f11a591/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/Pipeline.scala ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/Pipeline.scala b/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/Pipeline.scala deleted file mode 100644 index 65ab390..0000000 --- a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/Pipeline.scala +++ /dev/null @@ -1,27 +0,0 @@ -/** - * 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. - */ -package org.apache.eagle.stream.pipeline - - -import org.apache.eagle.stream.pipeline.compiler.PipelineCompiler -import org.apache.eagle.stream.pipeline.parser.PipelineParser -import org.apache.eagle.stream.pipeline.runner.PipelineRunner - -object Pipeline - extends PipelineRunner - with PipelineParser - with PipelineCompiler \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/0f11a591/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/annotation/Extension.scala ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/annotation/Extension.scala b/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/annotation/Extension.scala deleted file mode 100644 index 2ff81d4..0000000 --- a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/annotation/Extension.scala +++ /dev/null @@ -1,21 +0,0 @@ -/** - * 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. - */ -package org.apache.eagle.stream.pipeline.annotation - -import scala.annotation.StaticAnnotation - -case class Extension(extType:String) extends StaticAnnotation \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/0f11a591/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/compiler/PipelineCompiler.scala ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/compiler/PipelineCompiler.scala b/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/compiler/PipelineCompiler.scala deleted file mode 100644 index df8bbe5..0000000 --- a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/compiler/PipelineCompiler.scala +++ /dev/null @@ -1,61 +0,0 @@ -/** - * 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. - */ -package org.apache.eagle.stream.pipeline.compiler - - -import org.apache.eagle.datastream.core._ -import org.apache.eagle.stream.pipeline.extension.ModuleManager._ -import org.apache.eagle.stream.pipeline.parser._ -import org.apache.eagle.stream.pipeline.utils.CompileException - -trait PipelineCompiler { - def compile(pipeline:Pipeline):StreamContext = { - val context = new StreamContext(pipeline.config) - val dataflow = pipeline.dataflow - val dag = new StreamDAG(context.dag) - dataflow.getProcessors.map(buildStreamProducer(dag,_)).foreach(producer =>{ - producer.initWith(dag.graph,pipeline.config) - dag.addVertex(producer) - }) - dataflow.getConnectors.foreach(connector =>{ - val from = dag.getNodeByName(connector.from).get - val to = dag.getNodeByName(connector.to).get - dag.addEdge(from,to,buildStreamConnector(from,to,dataflow,connector)) - }) - context - } - private def buildStreamProducer(dag:StreamDAG,processor:Processor):StreamProducer[Any] = { - if(findModuleType(processor.getType)){ - getModuleMapperByType(processor.getType).map(processor).nameAs(processor.getId).stream(processor.streamId) - } else { - throw new CompileException(s"Unknown processor type [${processor.getType}]") - } - } - private def buildStreamConnector(from:StreamProducer[Any],to:StreamProducer[Any],dataflow:DataFlow,connector:Connector):StreamConnector[Any,Any]={ - var groupByIndexes:Seq[Int] = connector.groupByIndexes.orNull - if(groupByIndexes!=null ){ - if(connector.groupByFields.isDefined) throw new CompileException(s"Both ${Connector.GROUP_BY_FIELD_FIELD} and ${Connector.GROUP_BY_INDEX_FIELD} is defined at same time") - } else if(connector.groupByFields.isDefined){ - groupByIndexes = connector.groupByFields.get.map(dataflow.getProcessor(from.name).get.getSchema.get.indexOfAttribute) - } - if(groupByIndexes == null){ - ShuffleConnector(from,to) - } else { - GroupbyFieldsConnector(from,to,groupByIndexes) - } - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/0f11a591/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/extension/ModuleManager.scala ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/extension/ModuleManager.scala b/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/extension/ModuleManager.scala deleted file mode 100644 index 2174560..0000000 --- a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/extension/ModuleManager.scala +++ /dev/null @@ -1,186 +0,0 @@ -package org.apache.eagle.stream.pipeline.extension - -import java.util.Properties -import java.util.concurrent.atomic.AtomicBoolean - -import com.typesafe.config.ConfigFactory -import org.apache.eagle.dataproc.impl.storm.kafka.KafkaSourcedSpoutProvider -import org.apache.eagle.datastream.core._ -import org.apache.eagle.partition.PartitionStrategy -import org.apache.eagle.stream.pipeline.parser.Processor -import org.apache.kafka.clients.producer.{Callback, KafkaProducer, ProducerRecord, RecordMetadata} -//import org.slf4j.LoggerFactory - -import scala.collection.JavaConverters._ - -/** - * 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. - */ - - -object ModuleManager{ - def getModuleMapperByType(moduleType:String):ModuleMapper = { - classOfProcessorMapping(moduleType) - } - - def findModuleType(moduleType:String):Boolean = classOfProcessorMapping.contains(moduleType) - - val classOfProcessorMapping = Map[String,ModuleMapper]( - "KafkaSource" -> KafkaSourceStreamProducer, - "KafkaSink" -> KafkaSinkStreamProducer, - "Alert" -> AlertStreamProducer, - "Persistence" -> PersistProducer, - "Aggregator" -> AggregatorProducer, - "Console" -> ConsoleStreamProducer - ) -} - -trait ModuleMapper{ - def getType:String - def map(module:Processor):StreamProducer[Any] -} -object KafkaSourceStreamProducer extends ModuleMapper{ - def getType = "KafkaSource" - override def map(module:Processor): StreamProducer[Any] = { - val config = module.getConfig - new StormSourceProducer[Any](new KafkaSourcedSpoutProvider(null).getSpout(ConfigFactory.parseMap(config.asJava))) - } -} -object KafkaSinkStreamProducer extends ModuleMapper{ - def getType = "KafkaSink" - override def map(module:Processor): StreamProducer[Any] = { - val config = module.getConfig - ForeachProducer[AnyRef](KafkaSinkExecutor(config)) - } -} -object ConsoleStreamProducer extends ModuleMapper{ - override def getType: String = "Stdout" - override def map(module:Processor): StreamProducer[Any] = ForeachProducer[Any](m=>print(s"$m\n")) -} -object AlertStreamProducer extends ModuleMapper{ - def getType:String = "Alert" - override def map(module:Processor): StreamProducer[Any] = { - val config = module.getConfig - val moduleId = module.getId - // Support create functional AlertStreamProducer constructor - new AlertStreamProducer ( - upStreamNames = config.getOrElse("upStreamNames",if(module.inputIds!=null) module.inputIds.asJava else null).asInstanceOf[java.util.List[String]], - alertExecutorId = config.getOrElse("alertExecutorId",moduleId).asInstanceOf[String], - consume = config.getOrElse("consume",true).asInstanceOf[Boolean], - strategy = config.get("strategy") match {case Some(strategy)=> Class.forName(strategy.asInstanceOf[String]).newInstance().asInstanceOf[PartitionStrategy] case None => null} - ) - } -} - -object PersistProducer extends ModuleMapper{ - override def getType = "Persistence" - override def map(module:Processor): StreamProducer[Any] = { - val config = module.getConfig - new PersistProducer(config.getOrElse("executorId",module.getId).asInstanceOf[String],StorageType.withName(config.getOrElse("storageType",null).asInstanceOf[String])) - } -} - -object AggregatorProducer extends ModuleMapper{ - override def getType: String = "Aggregator" - override def map(module:Processor): StreamProducer[Any] = { - val config = module.getConfig - new AggregateProducer( - upStreamNames = config.getOrElse("upStreamNames",if(module.inputIds!=null) module.inputIds.asJava else null).asInstanceOf[java.util.List[String]], - config.getOrElse("analyzer",module.getId).asInstanceOf[String], - config.get("sql") match {case Some(sql) => sql.asInstanceOf[String] case None => null }, - config.get("strategy") match {case Some(strategy)=> Class.forName(strategy.asInstanceOf[String]).newInstance().asInstanceOf[PartitionStrategy] case None => null} - ) - } -} - -object KafkaSinkExecutor{ -// val LOG = LoggerFactory.getLogger(classOf[KafkaSinkExecutor]) -} - -/** - * @todo currently support single topic now, should support topic selector - * @param config - */ -case class KafkaSinkExecutor(config:Map[String,AnyRef]) extends ((AnyRef) => Unit) with Serializable{ - - val TOPIC_KEY = "topic" - def getDefaultProps = { - val props = new Properties() - props.putAll(Map[String,AnyRef]( - "bootstrap.servers" -> "localhost:6667", - "acks" -> "all", - "retries" -> "3", - "batch.size" -> "16384", - "linger.ms" -> "1", - "buffer.memory" -> "33554432", - "key.serializer" -> "org.apache.kafka.common.serialization.StringSerializer", - "value.serializer" -> classOf[org.apache.eagle.dataproc.impl.storm.kafka.JsonSerializer].getCanonicalName - ).asJava) - props - } - - @transient var initialized:AtomicBoolean = new AtomicBoolean(false) - @transient var producer:KafkaProducer[String,AnyRef] = null - @transient var topic:String = null - @transient var timeoutMs:Long = 3000 - - private def init():Unit = { - if(this.initialized != null && this.initialized.get()){ -// LOG.info("Already initialized, skip") - return - } - this.initialized = new AtomicBoolean(false) - if (producer != null) { -// LOG.info(s"Closing $producer") - producer.close() - } -// LOG.info("Initializing and creating Kafka Producer") - if (config.contains(TOPIC_KEY)) { - this.topic = config.get(TOPIC_KEY).get.asInstanceOf[String] - } else { - throw new IllegalStateException("topic is not defined") - } - val props = getDefaultProps - props.putAll((config - TOPIC_KEY).asJava) - producer = new KafkaProducer[String, AnyRef](props) -// LOG.info(s"Created new KafkaProducer: $producer") - initialized.set(true) - } - - override def apply(value: AnyRef): Unit = { - if(initialized == null || !initialized.get()) init() - if(topic == null) throw new IllegalStateException("topic is not defined") - val isList = value.isInstanceOf[java.util.List[AnyRef]] - val record: ProducerRecord[String, AnyRef] = if(isList){ - val list = value.asInstanceOf[java.util.List[AnyRef]] - if(list.size() == 1) { - new ProducerRecord[String, AnyRef](topic, value.asInstanceOf[java.util.List[AnyRef]].get(0)) - }else{ - new ProducerRecord[String, AnyRef](topic, value) - } - }else{ - new ProducerRecord[String, AnyRef](topic,value) - } - producer.send(record,new Callback(){ - override def onCompletion(metadata: RecordMetadata, exception: Exception): Unit = { - if(exception!=null){ -// LOG.error(s"Failed to send record $value to topic: $topic",exception) - throw new IllegalStateException(s"Failed to send record $value to topic: $topic",exception) - } - } - }) - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/0f11a591/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/parser/DataFlow.scala ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/parser/DataFlow.scala b/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/parser/DataFlow.scala deleted file mode 100644 index 7e1f4cf..0000000 --- a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/parser/DataFlow.scala +++ /dev/null @@ -1,235 +0,0 @@ -/** - * 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. - */ -package org.apache.eagle.stream.pipeline.parser - -import com.typesafe.config.Config -import org.apache.eagle.stream.pipeline.utils.ParseException - -import scala.collection.JavaConversions._ -import scala.collection.JavaConverters._ - -import scala.collection.mutable - - -class DataFlow { - def getInputs(id: String):Seq[Processor] = { - this.getConnectors.filter(_.to.equals(id)).map(c => getProcessor(c.from).get) - } - - /** - * Connect if not, do nothing if already connected - * - * @param from - * @param to - */ - def connect(from: String, to: String): Unit = { - val connector = Connector(from,to,null) - var exists = false - connectors.foreach(c => exists = (c.from.equals(from) && c.to.equals(to)) || exists) - if(!exists) addConnector(connector) - } - - private var processors = mutable.Map[String,Processor]() - private var connectors = mutable.Seq[Connector]() - def setProcessors(processors:Seq[Processor]):Unit = { - processors.foreach{module => - this.processors.put(module.getId,module) - } - } - def setProcessors(processors:mutable.Map[String,Processor]):Unit = { - this.processors = processors - } - def setConnectors(connectors:Seq[Connector]):Unit = { - connectors.foreach(connector =>{ - this.connectors :+= connector - }) - } - def addProcessor(module:Processor):Unit = { - if(contains(module)) throw new IllegalArgumentException(s"Duplicated processor id error, ${module.getId} has already been defined as ${getProcessor(module.getId)}") - processors.put(module.getId,module) - } - - def contains(module:Processor):Boolean = processors.contains(module.getId) - def addConnector(connector:Connector):Unit = { - connectors :+= connector - } - def getProcessors:Seq[Processor] = processors.values.toSeq - def getProcessor(processorId:String):Option[Processor] = processors.get(processorId) - def getConnectors:Seq[Connector] = connectors -} - -/** - * Stream Processor - * - * @param processorId - * @param processorType - * @param schema - * @param processorConfig - */ -case class Processor(var processorId:String = null,var processorType:String = null,var schema:Schema = null, var processorConfig:Map[String,AnyRef] = null) extends Serializable { - private[pipeline] var inputs:Seq[Processor] = null - private[pipeline] var inputIds:Seq[String] = null - - def getId:String = processorId - def getType:String = processorType - def getConfig:Map[String,AnyRef] = processorConfig - def getSchema:Option[Schema] = if(schema == null) None else Some(schema) - - /** - * @todo assume processorId as streamId - * @return - */ - def streamId = processorId -} - -case class Connector (from:String,to:String, config:Map[String,AnyRef]) extends Serializable{ - import Connector._ - - def group:Option[String] = config.get(GROUP_FIELD).asInstanceOf[Option[String]] - def groupByFields:Option[Seq[String]] = config.get(GROUP_BY_FIELD_FIELD) match { - case Some(obj) => Some(obj.asInstanceOf[java.util.List[String]].asScala.toSeq) - case None => None - } - def groupByIndexes:Option[Seq[Int]] = config.get(GROUP_BY_INDEX_FIELD) match { - case Some(obj) => Some(obj.asInstanceOf[java.util.List[java.lang.Integer]].asScala.toSeq.map(Int.unbox(_))) - case None => None - } -} - -object Connector{ - val GROUP_FIELD = "grouping" - val GROUP_BY_FIELD_FIELD = "groupByField" - val GROUP_BY_INDEX_FIELD = "groupByIndex" -} - -private [pipeline] -object Processor { - val SCHEMA_FIELD:String = "schema" - val INPUTS_FIELD = "inputs" - def parse(processorId:String,processorType:String,context:Map[String,AnyRef], schemaSet:SchemaSet):Processor = { - val schema = context.get(SCHEMA_FIELD) match { - case Some(schemaDef) => schemaDef match { - case schemaId:String => schemaSet.get(schemaId).getOrElse { - throw new ParseException(s"Schema [$schemaId] is not found but referred by [$processorType:$processorId] in $context") - } - case schemaMap:java.util.HashMap[String,AnyRef] => Schema.parse(schemaMap.toMap) - case _ => throw new ParseException(s"Illegal value for schema: $schemaDef") - } - case None => null - } - val instance = new Processor(processorId,processorType,schema,context-SCHEMA_FIELD) - if(context.contains(INPUTS_FIELD)) instance.inputIds = context.get(INPUTS_FIELD).get.asInstanceOf[java.util.List[String]].asScala.toSeq - instance - } -} - - -trait DataFlowParser { - def parse(config:Config,schemaSet:SchemaSet = SchemaSet.empty()):DataFlow = { - val dataw = new DataFlow() - val map = config.root().unwrapped().toMap - - // Parse processors and connectors - map.foreach(entry => { - parseSingle(entry._1,entry._2.asInstanceOf[java.util.HashMap[String,AnyRef]].toMap,dataw,schemaSet) - }) - expand(dataw) - validate(dataw) - dataw - } - - private def expand(datafw: DataFlow):Unit = { - datafw.getProcessors.foreach(proc =>{ - if(proc.inputIds!=null) { - proc.inputIds.foreach(id => { - // connect if not - datafw.connect(id,proc.getId) - }) - } - proc.inputs = datafw.getInputs(proc.getId) - proc.inputIds = proc.inputs.map(_.getId) - }) - } - - private def - validate(pipeline:DataFlow): Unit ={ - def checkModuleExists(id:String): Unit ={ - pipeline.getProcessor(id).orElse { - throw new ParseException(s"Stream [$id] is not defined before being referred") - } - } - - pipeline.getConnectors.foreach {connector => - checkModuleExists(connector.from) - checkModuleExists(connector.to) - } - } - - private def - parseSingle(identifier:String,config:Map[String,AnyRef],dataflow:DataFlow, schemaSet: SchemaSet):Unit = { - Identifier.parse(identifier) match { - case DefinitionIdentifier(processorType) => { - config foreach {entry => - dataflow.addProcessor(Processor.parse(entry._1, processorType,entry._2.asInstanceOf[java.util.HashMap[String, AnyRef]].toMap,schemaSet)) - } - } - case ConnectionIdentifier(fromIds,toId) => fromIds.foreach { fromId => - if(fromId.eq(toId)) throw new ParseException(s"Can't connect $fromId to $toId") - dataflow.addConnector(Connector(fromId,toId,config)) - } - case _ => ??? - } - } -} - - -private[pipeline] trait Identifier - -private[pipeline] case class DefinitionIdentifier(moduleType: String) extends Identifier -private[pipeline] case class ConnectionIdentifier(fromIds: Seq[String], toId: String) extends Identifier - -private[pipeline] object Identifier { - val ConnectorFlag = "->" - val UnitFlagSplitPattern = "\\|" - val UnitFlagChar = "|" - val ConnectorPattern = s"([\\w-|\\s]+)\\s+$ConnectorFlag\\s+([\\w-_]+)".r - def parse(identifier: String): Identifier = { - // ${id} -> ${id} - ConnectorPattern.findFirstMatchIn(identifier) match { - case Some(matcher) => { - if(matcher.groupCount != 2){ - throw new ParseException(s"Illegal connector definition: $identifier") - }else{ - val source = matcher.group(1) - val destination = matcher.group(2) - if(source.contains(UnitFlagChar)) { - val sources = source.split(UnitFlagSplitPattern).toSeq - ConnectionIdentifier(sources.map{_.trim()},destination) - }else{ - ConnectionIdentifier(Seq(source),destination) - } - } - } - case None => { - if(identifier.contains(ConnectorFlag)) throw new ParseException(s"Failed to parse $identifier") - DefinitionIdentifier(identifier) - } - } - } -} - -object DataFlow extends DataFlowParser \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/0f11a591/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/parser/Pipeline.scala ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/parser/Pipeline.scala b/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/parser/Pipeline.scala deleted file mode 100644 index eb09156..0000000 --- a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/parser/Pipeline.scala +++ /dev/null @@ -1,95 +0,0 @@ -package org.apache.eagle.stream.pipeline.parser - -import java.io.File - -import com.typesafe.config.{Config, ConfigFactory} - -/** - * 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. - */ - -case class Pipeline(config:Config,dataflow:DataFlow) - -/** - * Pipeline configuration parser - * - * For example: - * - * {{{ - * <code> - * { - * config { - * execution.environment.config = someValue - * } - * schema { - * metricStreamSchema { - * metric: string - * value: double - * timestamp: long - * } - * } - * dataflow { - * kafkaSource.source1 { - * schema = "metricStreamSchema" - * } - * kafkaSource.source2 { - * schema = { - * metric: string - * value: double - * timestamp: long - * } - * } - * } - * } - * </code> - * }}} - */ -trait PipelineParser{ - val CONFIG_FIELD = "config" - val SCHEMA_FIELD = "schema" - val DATAFLOW_FIELD = "dataflow" - - def parse(config:Config):Pipeline = { - if(config.isEmpty) throw new IllegalArgumentException("Pipeline configuration is empty") - var pConfig:Config = ConfigFactory.empty() - var pSchemaSet:SchemaSet = SchemaSet.empty() - var pDataflow:DataFlow = null - if(config.hasPath(CONFIG_FIELD)) pConfig = config.getConfig(CONFIG_FIELD) - if(config.hasPath(SCHEMA_FIELD)) pSchemaSet = SchemaSet.parse(config.getConfig(SCHEMA_FIELD)) - if(config.hasPath(DATAFLOW_FIELD)) pDataflow = DataFlow.parse(config.getConfig(DATAFLOW_FIELD),pSchemaSet) - - // Merge pipeline config over base config - val baseConfig =ConfigFactory.load() - pConfig = if(pConfig!=null) pConfig.withFallback(baseConfig) else baseConfig - new Pipeline(pConfig,pDataflow) - } - - def parseString(config:String):Pipeline = parse(ConfigFactory.parseString(config)) - - def parseStringWithConfig(dataFlow:String, config: Config) = { - val pConfig = config.withFallback(ConfigFactory.parseString(dataFlow)) - parse(pConfig) - } - - def parseResource(resource:String):Pipeline = { - // TODO: Load environment, currently hard-code with storm - if(resource.startsWith("/") || resource.startsWith("./")){ - parse(ConfigFactory.parseFile(new File(resource))) - } else{ - parse(ConfigFactory.parseResourcesAnySyntax(getClass.getClassLoader,resource)) - } - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/0f11a591/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/parser/Schema.scala ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/parser/Schema.scala b/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/parser/Schema.scala deleted file mode 100644 index 7653f9e..0000000 --- a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/parser/Schema.scala +++ /dev/null @@ -1,152 +0,0 @@ -/** - * 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. - */ -package org.apache.eagle.stream.pipeline.parser - -import com.typesafe.config.Config - -import scala.collection.JavaConversions.mapAsScalaMap -import scala.collection.mutable - -class Field(name:String) extends Serializable{ - def getName:String = name -} - -case class StringField(name:String) extends Field(name) -case class LongField(name:String) extends Field(name) -case class IntegerField(name:String) extends Field(name) -case class BooleanField(name:String) extends Field(name) -case class FloatField(name:String) extends Field(name) -case class DoubleField(name:String) extends Field(name) -case class DatetimeField(name:String,format:String) extends Field(name) - -object Field{ - def string(name:String) = StringField(name) - def long(name:String) = LongField(name) - def integer(name:String) = IntegerField(name) - def boolean(name:String) = BooleanField(name) - def float(name:String) = FloatField(name) - def double(name:String) = DoubleField(name) - def datetime(name:String)(format:String) = DatetimeField(name,format) - - def apply(name:String,typeName:String):Field = typeName match { - case "string" => string(name) - case "long" => long(name) - case "integer" => integer(name) - case "boolean" => boolean(name) - case "float" => float(name) - case "double" => double(name) - case _ => throw new UnsupportedOperationException(s"""Unknown attribute type $typeName for attribute "$name"""") - } -} - -case class Schema(attributes:Seq[Field]) extends Serializable{ - def getAttribute(attributeName:String):Option[Field]={ - if(attributes != null){ - attributes.find(_.getName.eq(attributeName)) - }else None - } - - def indexOfAttribute(attributeName:String):Int = { - if(attributes != null){ - attributes.indexWhere(_.getName.eq(attributeName)) - } else -1 - } - - @throws[IllegalArgumentException] - def indexOfAttributeOrException(attributeName:String):Int = { - if(attributes != null){ - attributes.indexWhere(_.getName.eq(attributeName)) - } else throw new IllegalArgumentException(s"Attribute [$attributeName] is not found in stream $this") - } -} - -object Schema{ - def parse(map:Map[String,AnyRef]):Schema = { - new Schema(map.keys.map {attributeName => - map(attributeName) match{ - case simpleType:String => Field(attributeName,simpleType) - case complexType:java.util.Map[String,AnyRef] => throw new IllegalStateException(s"ComplexType attribute definition is not supported yet [$attributeName : $complexType] ") - case otherType@_ => throw new IllegalStateException(s"Illegal attribute definition $attributeName : $otherType") - } - }.toSeq) - } - - /** - * @param attributes support string, symbol, Attribute and so on. - * @return - */ - def build(attributes:Seq[AnyRef]):Schema = { - new Schema(attributes.map{ a:AnyRef => - a match { - case t:(String, AnyRef) => { - t._2 match { - case v:String => Field(t._1,v) - case v:Symbol => Field(t._1,v.name) - case _ => throw new UnsupportedOperationException(s"Illegal attribute definition $a") - } - } - case t:Field => t - case _ => throw new UnsupportedOperationException(s"Illegal attribute definition $a") - } - }) - } -} - -private[pipeline] class StreamUndefinedException(message:String = "stream is not defined",throwable: Throwable = null) extends Exception(message,throwable) - -private[pipeline] class SchemaSet { - private val processorSchemaCache = mutable.Map[String,Schema]() - def set(schemaId:String,schema:Schema):Unit = { - if(processorSchemaCache.contains(schemaId)) throw new IllegalArgumentException( - s""" - |Failed to define schema for $schemaId as $schema, - |because it has been defined as ${processorSchemaCache(schemaId)}, - |please call updateSchema(processorId,schema) instead - """) - processorSchemaCache.put(schemaId,schema) - } - def get(schemaId:String):Option[Schema] = processorSchemaCache.get(schemaId) -} - -private[pipeline] object SchemaSet{ - def empty() = new SchemaSet() - /** - * For example: - * - * <code> - * { - * metricStream { - * metric: string - * value: double - * timestamp: long - * } - * } - * </code> - * @param schemaConfig - * @return - */ - def parse(schemaConfig:Map[String,AnyRef]):SchemaSet = { - val schemas = new SchemaSet() - schemaConfig.foreach(entry =>{ - schemas.set(entry._1,Schema.parse(entry._2.asInstanceOf[java.util.HashMap[String,AnyRef]].toMap)) - }) - schemas - } - - def parse(config:Config):SchemaSet = parse(config.root().unwrapped().asInstanceOf[java.util.HashMap[String,AnyRef]].toMap) -} - http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/0f11a591/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/runner/PipelineRunner.scala ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/runner/PipelineRunner.scala b/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/runner/PipelineRunner.scala deleted file mode 100644 index 1c964e1..0000000 --- a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/runner/PipelineRunner.scala +++ /dev/null @@ -1,115 +0,0 @@ -/** - * 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. - */ -package org.apache.eagle.stream.pipeline.runner - -import java.util - -import com.typesafe.config.Config -import org.apache.commons.cli.{CommandLine, Options} -import org.apache.eagle.dataproc.util.ConfigOptionParser -import org.apache.eagle.datastream.ExecutionEnvironments.storm -import org.apache.eagle.datastream.core.ExecutionEnvironment -import org.apache.eagle.stream.pipeline.compiler.PipelineCompiler -import org.apache.eagle.stream.pipeline.parser.PipelineParser -import org.slf4j.LoggerFactory - -import scala.reflect.runtime.{universe => ru} - -trait PipelineRunner extends PipelineParser with PipelineCompiler{ - import PipelineCLIOptionParser._ - private val LOG = LoggerFactory.getLogger("PipelineCLIOptionParser") - def submit[T <: ExecutionEnvironment](resource:String)(implicit typeTag:ru.TypeTag[T]) = - compile(parseResource(resource)).submit[T] - def submit(resource:String,clazz:Class[ExecutionEnvironment]) = - compile(parseResource(resource)).submit(clazz) - def submit(pipelineConfig:Config,clazz:Class[ExecutionEnvironment]) = - compile(parse(pipelineConfig)).submit(clazz) - def submit[T <: ExecutionEnvironment](pipelineConfig:Config)(implicit typeTag: ru.TypeTag[T]) = - compile(parse(pipelineConfig)).submit[T] - - def apply(args:Array[String]):PipelineRunner = { - new ConfigOptionParser().load(args) - this - } - - def main(args: Array[String]): Unit = { - val config = PipelineCLIOptionParser.load(args) - if(config.hasPath(PIPELINE_CONFIG_KEY)) { - submit[storm](config.getString(PIPELINE_CONFIG_KEY)) - } else { - sys.error( - s""" - |Error: --$PIPELINE_OPT_KEY is required - |$USAGE - """.stripMargin) - } - } -} - -private[runner] object PipelineCLIOptionParser extends ConfigOptionParser{ - val LOG = LoggerFactory.getLogger("PipelineCLIOptionParser") - val PIPELINE_OPT_KEY="pipeline" - - val PIPELINE_CONFIG_KEY="pipeline.config" - - val CONFIG_OPT_KEY="conf" - val CONFIG_RESOURCE_KEY="config.resource" - val CONFIG_FILE_KEY="config.file" - val USAGE = - """ - |Usage: java org.apache.eagle.stream.pipeline.Pipeline [options] - | - |Options: - | --pipeline pipeline configuration - | --conf common configuration - | --env storm (support spark, etc later) - | --mode local/remote/cluster - """.stripMargin - - override protected def options(): Options = { - val options = super.options() - options.addOption(PIPELINE_OPT_KEY, true, "Pipeline configuration file") - options.addOption(CONFIG_OPT_KEY, true, "Config properties file") - options - } - - override protected def parseCommand(cmd: CommandLine): util.Map[String, String] = { - val map = super.parseCommand(cmd) - - if (cmd.hasOption(PIPELINE_OPT_KEY)) { - val pipelineConf = cmd.getOptionValue(PIPELINE_OPT_KEY) - if(pipelineConf == null){ - throw new IllegalArgumentException(s"--$PIPELINE_OPT_KEY should not be null") - } else { - LOG.info(s"Set $PIPELINE_CONFIG_KEY as $pipelineConf") - map.put(PIPELINE_CONFIG_KEY, pipelineConf) - } - } - - if(cmd.hasOption(CONFIG_OPT_KEY)){ - val commonConf = cmd.getOptionValue(CONFIG_OPT_KEY) - if(commonConf.contains("/")){ - LOG.info(s"Set $CONFIG_FILE_KEY as $commonConf") - map.put(CONFIG_FILE_KEY, commonConf) - }else { - LOG.info(s"Set $CONFIG_RESOURCE_KEY $commonConf") - map.put(CONFIG_RESOURCE_KEY, commonConf) - } - } - map - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/0f11a591/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/utils/Exceptions.scala ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/utils/Exceptions.scala b/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/utils/Exceptions.scala deleted file mode 100644 index 1102a33..0000000 --- a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/main/scala/org/apache/eagle/stream/pipeline/utils/Exceptions.scala +++ /dev/null @@ -1,20 +0,0 @@ -/** - * 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. - */ -package org.apache.eagle.stream.pipeline.utils - -class ParseException(message:String) extends Exception(message) -class CompileException(message:String) extends Exception(message) \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/0f11a591/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/application.conf ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/application.conf b/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/application.conf deleted file mode 100644 index 3e8f69c..0000000 --- a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/application.conf +++ /dev/null @@ -1,34 +0,0 @@ -# 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. - -{ - "eagleProps" : { - "dataJoinPollIntervalSec" : 30 - "mailHost" : "smtp.server.host" - "mailSmtpPort":"25" - "mailDebug" : "true" - "eagleService": { - "host": "localhost" - "port": 9099 - "username": "admin" - "password": "secret" - } - } - "dynamicConfigSource" : { - "enabled" : true - "initDelayMillis" : 0 - "delayMillis" : 30000 - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/0f11a591/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/eagle-pipeline.sh ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/eagle-pipeline.sh b/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/eagle-pipeline.sh deleted file mode 100644 index 4250681..0000000 --- a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/eagle-pipeline.sh +++ /dev/null @@ -1,20 +0,0 @@ -#!/usr/bin/env bash -# 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. - -# ./eagle-pipeline.sh --pipeline [pipeline-definition-config] --config [base-configuration] - - - http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/0f11a591/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/log4j.properties ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/log4j.properties b/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/log4j.properties deleted file mode 100644 index c8a4f46..0000000 --- a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/log4j.properties +++ /dev/null @@ -1,19 +0,0 @@ -# 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. - -log4j.rootLogger=INFO, stdout -log4j.appender.stdout=org.apache.log4j.ConsoleAppender -log4j.appender.stdout.layout=org.apache.log4j.PatternLayout -log4j.appender.stdout.layout.ConversionPattern=%d{ISO8601} %p [%t] %c{2}[%L]: %m%n \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/0f11a591/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/pipeline_1.conf ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/pipeline_1.conf b/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/pipeline_1.conf deleted file mode 100644 index 6dddf7a..0000000 --- a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/pipeline_1.conf +++ /dev/null @@ -1,131 +0,0 @@ -# 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. - -{ - config { - alertExecutorConfigs { - defaultAlertExecutor { - "parallelism" : 1 - "partitioner" : "org.apache.eagle.policy.DefaultPolicyPartitioner" - "needValidation" : "true" - } - } - eagleProps { - "site" : "sandbox" - "application": "eventSource" - "dataJoinPollIntervalSec" : 30 - "mailHost" : "mail.host.com" - "mailSmtpPort":"25" - "mailDebug" : "true" - "eagleService": { - "host": "localhost" - "port": 38080 - "username": "admin" - "password": "secret" - } - } - dynamicConfigSource { - "enabled" : true - "initDelayMillis" : 0 - "delayMillis" : 30000 - } - } - - schema { - metricStreamSchema { - metric: string - value: double - timestamp: long - } - } - - dataflow { - KafkaSource.metricStream_1 { - parallism = 1000 - topic = "metric_event_1" - zkConnection = "localhost:2181" - zkConnectionTimeoutMS = 15000 - consumerGroupId = "Consumer" - fetchSize = 1048586 - transactionZKServers = "localhost" - transactionZKPort = 2181 - transactionZKRoot = "/consumers" - transactionStateUpdateMS = 2000 - deserializerClass = "org.apache.eagle.datastream.storm.JsonMessageDeserializer" - schema = "metricStreamSchema" - } - - KafkaSource.metricStream_2 { - parallism = 1000 - topic = "metric_event_2" - zkConnection = "localhost:2181" - zkConnectionTimeoutMS = 15000 - consumerGroupId = "Consumer" - fetchSize = 1048586 - transactionZKServers = "localhost" - transactionZKPort = 2181 - transactionZKRoot = "/consumers" - transactionStateUpdateMS = 2000 - deserializerClass = "org.apache.eagle.datastream.storm.JsonMessageDeserializer" - } - - KafkaSource.metricStream_3{ - parallism = 1000 - topic = "metric_event_3" - zkConnection = "localhost:2181" - zkConnectionTimeoutMS = 15000 - consumerGroupId = "Consumer" - fetchSize = 1048586 - transactionZKServers = "localhost" - transactionZKPort = 2181 - transactionZKRoot = "/consumers" - transactionStateUpdateMS = 2000 - deserializerClass = "org.apache.eagle.datastream.storm.JsonMessageDeserializer" - schema = "metricStreamSchema" - } - - KafkaSink.metricStore { - schema = "metricStreamSchema" - parallism = 1000 - topic = "metric_event_2" - zkConnection = "localhost:2181" - zkConnectionTimeoutMS = 15000 - consumerGroupId = "Consumer" - fetchSize = 1048586 - transactionZKServers = "localhost" - transactionZKPort = 2181 - transactionZKRoot = "/consumers" - transactionStateUpdateMS = 2000 - deserializerClass = "org.apache.eagle.datastream.storm.JsonMessageDeserializer" - } - - Alert.alert { -// upStreamNames = [metricStream_1,metricStream_2] - alertExecutorId = defaultAlertExecutor - } - -// aggregator.aggreator { -// executor = "aggreationExecutor" -// } - - metricStream_1|metricStream_2 -> alert { - group = shuffle - } - - metricStream_1|metricStream_2 -> metricStore { - group = shuffle - } - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/0f11a591/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/pipeline_2.conf ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/pipeline_2.conf b/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/pipeline_2.conf deleted file mode 100644 index 5e3561a..0000000 --- a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/pipeline_2.conf +++ /dev/null @@ -1,93 +0,0 @@ -# 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. - -{ - config { - envContextConfig { - "env" : "storm" - "mode" : "local" - "topologyName" : "dsl-based-topology" - } - eagleProps { - "site" : "sandbox" - "application": "eventSource" - "dataJoinPollIntervalSec" : 30 - "mailHost" : "mail.host.com" - "mailSmtpPort":"25" - "mailDebug" : "true" - "eagleService": { - "host": "localhost" - "port": 38080 - "username": "admin" - "password": "secret" - } - } - dynamicConfigSource { - "enabled" : true - "initDelayMillis" : 0 - "delayMillis" : 30000 - } - } - - dataflow { - KafkaSource.metricStream_1 { - parallism = 1000 - topic = "metric_event_1" - zkConnection = "localhost:2181" - zkConnectionTimeoutMS = 15000 - consumerGroupId = "Consumer" - fetchSize = 1048586 - transactionZKServers = "localhost" - transactionZKPort = 2181 - transactionZKRoot = "/consumers" - transactionStateUpdateMS = 2000 - deserializerClass = "org.apache.eagle.datastream.storm.JsonMessageDeserializer" - } - - KafkaSource.metricStream_2 { - parallism = 1000 - topic = "metric_event_2" - zkConnection = "localhost:2181" - zkConnectionTimeoutMS = 15000 - consumerGroupId = "Consumer" - fetchSize = 1048586 - transactionZKServers = "localhost" - transactionZKPort = 2181 - transactionZKRoot = "/consumers" - transactionStateUpdateMS = 2000 - deserializerClass = "org.apache.eagle.datastream.storm.JsonMessageDeserializer" - } - - KafkaSource.metricStream_3{ - parallism = 1000 - topic = "metric_event_3" - zkConnection = "localhost:2181" - zkConnectionTimeoutMS = 15000 - consumerGroupId = "Consumer" - fetchSize = 1048586 - transactionZKServers = "localhost" - transactionZKPort = 2181 - transactionZKRoot = "/consumers" - transactionStateUpdateMS = 2000 - deserializerClass = "org.apache.eagle.datastream.storm.JsonMessageDeserializer" - } - - Console.printer {} - - metricStream_1|metricStream_2|metricStream_3 -> printer { - grouping = shuffle - } - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/0f11a591/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/pipeline_3.conf ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/pipeline_3.conf b/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/pipeline_3.conf deleted file mode 100644 index 9dc7ce3..0000000 --- a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/pipeline_3.conf +++ /dev/null @@ -1,152 +0,0 @@ -# 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. - -{ - config { - envContextConfig { - "env" : "storm" - "mode" : "local" - "topologyName" : "dsl-based-topology" - "parallelismConfig" : { - "kafkaMsgConsumer" : 1 - } - } - alertExecutorConfigs { - defaultAlertExecutor { - "parallelism" : 1 - "partitioner" : "org.apache.eagle.policy.DefaultPolicyPartitioner" - "needValidation" : "true" - } - } - eagleProps { - "site" : "sandbox" - "application": "HADOOP" - "dataJoinPollIntervalSec" : 30 - "mailHost" : "some.mail.server" - "mailSmtpPort":"25" - "mailDebug" : "true" - "eagleService": { - "host": "localhost" - "port": 38080 - "username": "admin" - "password": "secret" - } - } - dynamicConfigSource { - "enabled" : true - "initDelayMillis" : 0 - "delayMillis" : 30000 - } - } - - schema { -// JmxStreamOne { -// attributes { -// metric: string -// value: double -// timestamp: long -// } -// alertExecutorId = [defaultAlertExecutor,anotherAlertExecutor] -// } - JmxStreamOne { - metric: string - value: double - timestamp: long - } - } - - dataflow { - KafkaSource.JmxStreamOne { - parallism = 1000 - topic = "metric_event_1" - zkConnection = "localhost:2181" - zkConnectionTimeoutMS = 15000 - consumerGroupId = "Consumer" - fetchSize = 1048586 - transactionZKServers = "localhost" - transactionZKPort = 2181 - transactionZKRoot = "/consumers" - transactionStateUpdateMS = 2000 - deserializerClass = "org.apache.eagle.datastream.storm.JsonMessageDeserializer" - } - - KafkaSource.JmxStreamTwo { - parallism = 1000 - topic = "metric_event_2" - zkConnection = "localhost:2181" - zkConnectionTimeoutMS = 15000 - consumerGroupId = "Consumer" - fetchSize = 1048586 - transactionZKServers = "localhost" - transactionZKPort = 2181 - transactionZKRoot = "/consumers" - transactionStateUpdateMS = 2000 - deserializerClass = "org.apache.eagle.datastream.storm.JsonMessageDeserializer" - } - - KafkaSource.JmxStreamThree{ - parallism = 1000 - topic = "metric_event_3" - zkConnection = "localhost:2181" - zkConnectionTimeoutMS = 15000 - consumerGroupId = "Consumer" - fetchSize = 1048586 - transactionZKServers = "localhost" - transactionZKPort = 2181 - transactionZKRoot = "/consumers" - transactionStateUpdateMS = 2000 - deserializerClass = "org.apache.eagle.datastream.storm.JsonMessageDeserializer" - } - - Console.printer { - format = "%s" - } - - KafkaSink.metricStore { - topic = "metric_event_persist" - } - -// KafkaSink.alertStore { -// "topic" = "alert_persist" -// "bootstrap.servers" = "localhost:6667" -// } - - Alert.alert { - inputs = [JmxStreamOne,JmxStreamTwo,JmxStreamThree] - - upStreamNames = [JmxStreamOne,JmxStreamTwo,JmxStreamThree] - alertExecutorId = defaultAlertExecutor - } - -// Aggregator.aggreator { -// upStreamNames = [] -// analyzerId = "" -// cepQl = "" -// strategy = "" -// } - - JmxStreamOne|JmxStreamTwo|JmxStreamThree -> alert { - grouping = shuffle - } - - JmxStreamOne|JmxStreamTwo|JmxStreamThree -> metricStore { - grouping = shuffle - } - - JmxStreamOne|JmxStreamTwo|JmxStreamThree -> printer { - grouping = shuffle - } - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/0f11a591/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/pipeline_4.conf ---------------------------------------------------------------------- diff --git a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/pipeline_4.conf b/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/pipeline_4.conf deleted file mode 100644 index 9c35456..0000000 --- a/eagle-core/eagle-data-process/eagle-stream-pipeline/src/test/resources/pipeline_4.conf +++ /dev/null @@ -1,125 +0,0 @@ -# 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. - -{ - config { - envContextConfig { - "env" : "storm" - "mode" : "local" - "topologyName" : "dsl-topology" - "parallelismConfig" : { - "kafkaMsgConsumer" : 1 - }, - } - alertExecutorConfigs { - defaultAlertExecutor { - "parallelism" : 1 - "partitioner" : "org.apache.eagle.policy.DefaultPolicyPartitioner" - "needValidation" : "true" - } - } - eagleProps { - "site" : "sandbox" - "application": "HADOOP" - } - } - - dataflow { - KafkaSource.JmxStreamOne { - parallism = 1000 - topic = "metric_event_1" - zkConnection = "localhost:2181" - zkConnectionTimeoutMS = 15000 - consumerGroupId = "Consumer" - fetchSize = 1048586 - transactionZKServers = "localhost" - transactionZKPort = 2181 - transactionZKRoot = "/consumers" - transactionStateUpdateMS = 2000 - deserializerClass = "org.apache.eagle.datastream.storm.JsonMessageDeserializer" - } - - KafkaSource.JmxStreamTwo { - parallism = 1000 - topic = "metric_event_2" - zkConnection = "localhost:2181" - zkConnectionTimeoutMS = 15000 - consumerGroupId = "Consumer" - fetchSize = 1048586 - transactionZKServers = "localhost" - transactionZKPort = 2181 - transactionZKRoot = "/consumers" - transactionStateUpdateMS = 2000 - deserializerClass = "org.apache.eagle.datastream.storm.JsonMessageDeserializer" - } - - KafkaSource.JmxStreamThree{ - parallism = 1000 - topic = "metric_event_3" - zkConnection = "localhost:2181" - zkConnectionTimeoutMS = 15000 - consumerGroupId = "Consumer" - fetchSize = 1048586 - transactionZKServers = "localhost" - transactionZKPort = 2181 - transactionZKRoot = "/consumers" - transactionStateUpdateMS = 2000 - deserializerClass = "org.apache.eagle.datastream.storm.JsonMessageDeserializer" - } - - Console.printer { - format = "%s" - } - - KafkaSink.metricStore { - topic = "metric_event_persist" - } - -// KafkaSink.aggSink { -// topic = "metric_agg_persist" -// } - - Alert.defaultAlertExecutor { - // upStreamNames = [JmxStreamOne,JmxStreamTwo,JmxStreamThree] - // alertExecutorId = defaultAlertExecutor - } - - Aggregator.Aggregator{ sql = """ - define stream JmxStreamOne(eagleAlertContext object, timestamp long, metric string, value double); - @info(name = "query") - from JmxStreamOne[value > 100.0] select * insert into outputStream; - """} - - - JmxStreamOne -> Aggregator {} - - Aggregator -> printer {} - -// Aggregator -> aggregatedSink{ -// grouping = shuffle -// } - JmxStreamOne|JmxStreamTwo|JmxStreamThree -> defaultAlertExecutor { - grouping = shuffle - } - - JmxStreamOne|JmxStreamTwo|JmxStreamThree -> metricStore { - grouping = shuffle - } - - JmxStreamOne|JmxStreamTwo|JmxStreamThree -> printer { - grouping = shuffle - } - } -} \ No newline at end of file
