diff --git a/app/java/src/main/java/io/qiniu/StartUpApplicationRunner.java b/app/java/src/main/java/io/qiniu/StartUpApplicationRunner.java index 4f45b50..dfa2afe 100644 --- a/app/java/src/main/java/io/qiniu/StartUpApplicationRunner.java +++ b/app/java/src/main/java/io/qiniu/StartUpApplicationRunner.java @@ -1,10 +1,11 @@ package io.qiniu; import com.qiniu.pandora.common.QiniuException; +import com.qiniu.pandora.util.StringUtils; import io.qiniu.configuration.PandoraProperties; import io.qiniu.service.PandoraService; -import java.io.BufferedReader; import java.io.BufferedWriter; +import java.io.File; import java.io.IOException; import java.io.RandomAccessFile; import java.nio.channels.FileChannel; @@ -20,6 +21,7 @@ import org.springframework.boot.ApplicationRunner; import org.springframework.core.annotation.Order; import org.springframework.stereotype.Component; +import org.springframework.util.ObjectUtils; @Component @Order(1) @@ -94,11 +96,9 @@ private void setId() throws IOException { Path idPath = Paths.get(ID_PATH); String id = properties.getId(); - BufferedReader reader = Files.newBufferedReader(idPath); - String content = reader.readLine(); - - if (id != null) { - content = id; + File file = idPath.toFile(); + if (file.exists() && ObjectUtils.isEmpty(id)) { + id = StringUtils.utf8String(Files.readAllBytes(idPath)); } // generate a new id and save it to file @@ -112,7 +112,7 @@ private void setId() throws IOException { properties.getServerAddress(), properties.getServerPort(), properties.getPandoraToken(), - content); + id); if (id == null) { throw new QiniuException("start failed, worker id is null"); diff --git a/app/java/src/main/java/io/qiniu/common/entity/collector/CollectorTask.java b/app/java/src/main/java/io/qiniu/common/entity/collector/CollectorTask.java index 3fabe11..fb08cc7 100644 --- a/app/java/src/main/java/io/qiniu/common/entity/collector/CollectorTask.java +++ b/app/java/src/main/java/io/qiniu/common/entity/collector/CollectorTask.java @@ -26,7 +26,7 @@ public class CollectorTask { private Map config; // collector config - private Map meta; // ex. runner meta data + private String meta; // ex. runner meta data private long createTime; @@ -61,7 +61,7 @@ public CollectorTask( boolean enabled, boolean status, Map config, - Map meta, + String meta, long createTime, long updateTime) { this.id = id; @@ -174,11 +174,11 @@ public void setConfig(Map config) { this.config = config; } - public Map getMeta() { + public String getMeta() { return meta; } - public void setMeta(Map meta) { + public void setMeta(String meta) { this.meta = meta; } @@ -237,7 +237,7 @@ public CollectorTask toCollectorTask() throws ParseException { task.setStatus(status == 1); task.setConfig(JsonHelper.readValueAsMap(config.getBytes())); if (this.meta != null) { - task.setMeta(JsonHelper.readValueAsMap(meta.getBytes())); + task.setMeta(meta); } task.setCreateTime(df.parse(createTime).getTime()); task.setUpdateTime(df.parse(updateTime).getTime()); diff --git a/app/java/src/main/java/io/qiniu/configuration/CollectorProperties.java b/app/java/src/main/java/io/qiniu/configuration/CollectorProperties.java new file mode 100644 index 0000000..226c232 --- /dev/null +++ b/app/java/src/main/java/io/qiniu/configuration/CollectorProperties.java @@ -0,0 +1,19 @@ +package io.qiniu.configuration; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.context.annotation.Configuration; +import org.springframework.stereotype.Component; + +@Component +@Configuration +@ConfigurationProperties(prefix = "collector") +public class CollectorProperties { + + @Value("${collector.meta_path:}") + private String metaPath; + + public String getMetaPath() { + return metaPath; + } +} diff --git a/app/java/src/main/java/io/qiniu/configuration/PandoraProperties.java b/app/java/src/main/java/io/qiniu/configuration/PandoraProperties.java index 8125659..5494f7b 100644 --- a/app/java/src/main/java/io/qiniu/configuration/PandoraProperties.java +++ b/app/java/src/main/java/io/qiniu/configuration/PandoraProperties.java @@ -1,5 +1,6 @@ package io.qiniu.configuration; +import io.qiniu.common.Constant; import io.qiniu.common.entity.pandora.PandoraMode; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.context.properties.ConfigurationProperties; @@ -27,10 +28,10 @@ public class PandoraProperties { @Value("${server.port:8088}") private String serverPort; - @Value("${pandora.app:demo}") + @Value("${pandora.app:" + Constant.APP_NAME + "}") private String appName; - @Value("${pandora.service:demo}") + @Value("${pandora.service:" + Constant.APP_NAME + "}") private String serviceName; @Value("${pandora.id:}") diff --git a/app/java/src/main/java/io/qiniu/service/PandoraService.java b/app/java/src/main/java/io/qiniu/service/PandoraService.java index f26676a..d8077a8 100644 --- a/app/java/src/main/java/io/qiniu/service/PandoraService.java +++ b/app/java/src/main/java/io/qiniu/service/PandoraService.java @@ -3,6 +3,8 @@ import com.qiniu.pandora.DefaultPandoraClient; import com.qiniu.pandora.service.customservice.CustomService; import com.qiniu.pandora.service.storage.StorageService; +import com.qiniu.pandora.service.token.TokenService; +import com.qiniu.pandora.service.upload.PostDataService; import io.qiniu.configuration.PandoraProperties; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; @@ -13,12 +15,16 @@ public class PandoraService { private final DefaultPandoraClient client; private final CustomService customService; private final StorageService storageService; + private final TokenService tokenService; + private final PostDataService postDataService; @Autowired public PandoraService(PandoraProperties properties) { this.client = new DefaultPandoraClient(properties.getPandoraUrl()); this.customService = client.NewCustomService(); this.storageService = client.NewStorageService(); + this.tokenService = client.NewTokenService(); + this.postDataService = client.NewPostDataService(tokenService, properties.getPandoraToken()); } public DefaultPandoraClient getClient() { @@ -32,4 +38,12 @@ public CustomService getCustomService() { public StorageService getStorageService() { return storageService; } + + public TokenService getTokenService() { + return tokenService; + } + + public PostDataService getPostDataService() { + return postDataService; + } } diff --git a/app/java/src/main/java/io/qiniu/service/collector/CollectorTaskService.java b/app/java/src/main/java/io/qiniu/service/collector/CollectorTaskService.java index 205c520..3e50d67 100644 --- a/app/java/src/main/java/io/qiniu/service/collector/CollectorTaskService.java +++ b/app/java/src/main/java/io/qiniu/service/collector/CollectorTaskService.java @@ -1,14 +1,18 @@ package io.qiniu.service.collector; import com.qiniu.pandora.collect.Collector; +import com.qiniu.pandora.collect.CollectorConfig; +import com.qiniu.pandora.collect.CollectorContext; import com.qiniu.pandora.collect.DefaultCollector; import com.qiniu.pandora.collect.State; -import com.qiniu.pandora.collect.runner.config.CollectorConfig; +import com.qiniu.pandora.collect.runner.config.RunnerConfig; import com.qiniu.pandora.common.QiniuException; import com.qiniu.pandora.common.QiniuExceptions; import io.qiniu.common.entity.collector.CollectorTask; +import io.qiniu.configuration.CollectorProperties; import io.qiniu.configuration.PandoraProperties; import io.qiniu.dao.collector.ICollectorTaskDao; +import io.qiniu.service.PandoraService; import java.util.ArrayList; import java.util.HashMap; import java.util.List; @@ -33,13 +37,17 @@ public class CollectorTaskService { @Autowired public CollectorTaskService( PandoraProperties properties, - ICollectorTaskDao collectorTaskDao, - PandoraProperties pandoraProperties) { + CollectorProperties collectorProperties, + PandoraService pandoraService, + ICollectorTaskDao collectorTaskDao) { this.workerId = properties.getId(); this.collectorTaskDao = collectorTaskDao; - this.properties = pandoraProperties; - - this.collector = new DefaultCollector(); + this.properties = properties; + this.collector = + new DefaultCollector( + new CollectorConfig(collectorProperties.getMetaPath()), + new CollectorContext( + pandoraService.getTokenService(), pandoraService.getPostDataService())); } @PostConstruct @@ -50,7 +58,7 @@ public void init() { public List queryTasks() throws QiniuException { List tasks = new ArrayList<>(); - for (CollectorConfig config : this.collector.getAllRunners()) { + for (RunnerConfig config : this.collector.getAllRunners()) { CollectorTask task = convertConfigToTask(config); task.setWorkerId(this.workerId); tasks.add(task); @@ -60,7 +68,7 @@ public List queryTasks() throws QiniuException { public List queryTasks(List taskIds) throws QiniuException { List tasks = new ArrayList<>(); - for (CollectorConfig config : this.collector.getRunners(taskIds)) { + for (RunnerConfig config : this.collector.getRunners(taskIds)) { CollectorTask task = convertConfigToTask(config); task.setWorkerId(this.workerId); tasks.add(task); @@ -69,7 +77,7 @@ public List queryTasks(List taskIds) throws QiniuExceptio } public CollectorTask queryTask(String taskId) throws QiniuException { - CollectorConfig config = this.collector.getRunner(taskId); + RunnerConfig config = this.collector.getRunner(taskId); if (config == null) { return null; } @@ -173,15 +181,16 @@ private void recoveryTask() { } } - private static CollectorConfig convertTaskToConfig(CollectorTask task) throws QiniuException { - return new CollectorConfig( + private static RunnerConfig convertTaskToConfig(CollectorTask task) throws QiniuException { + return new RunnerConfig( task.getTaskId(), task.getName(), task.isEnabled() ? State.STARTED : State.STOPPED, + task.getMeta(), convertConfigToProperties(task.getConfig())); } - private static CollectorTask convertConfigToTask(CollectorConfig config) throws QiniuException { + private static CollectorTask convertConfigToTask(RunnerConfig config) throws QiniuException { CollectorTask task = new CollectorTask(); task.setTaskId(config.getId()); diff --git a/sdk/README.md b/sdk/README.md new file mode 100644 index 0000000..aca61d4 --- /dev/null +++ b/sdk/README.md @@ -0,0 +1,46 @@ +Pandora-Java-SDK +============= + +提供访问Pandora平台所需的接口以及服务端采集基础框架 + +平台接口 +============= + +1. app相关 +2. 元数据管理 +3. Token管理 +4. 数据传输 + +服务端采集框架 +============= +改造自flume EmbeddedAgent框架： +1. 增加自定义source、sink +2. 增加source meta管理 + +采集配置样例： +```json +{ + "source.type":"file", + "source.input.file":"test.log", + "channel.type":"memory", + "channel.capacity": "200", + "sinks":"sink1", + "sink1.type":"pandora", + "sink1.source_type": "json", + "sink1.repo": "test", + "processor.type": "default" +} +``` + +采集配置介绍： +1. source数据来源: file + 1. file: 文件按行采集，暂不支持增量采集 +2. channel缓存管道: memory, file + 1. memory: 内存管道，数据暂存内存中，性能较高但程序崩溃时会导致数据丢失 + 2. file: 使用本地文件作为缓存管道 +3. sink发送源: pandora + 1. pandora: 将数据发往pandora平台 +4. processor类型: default、load_balance、failover + 1. default: 默认类型，不支持多sink。 + 2. load_balance: 负载均衡，有round_robin(轮询)或random(随机)两种选择机制，默认round_robin。 + 3. failover: 维护sink优先级列表，支持故障转移 \ No newline at end of file diff --git a/sdk/pom.xml b/sdk/pom.xml index 617740a..463c471 100644 --- a/sdk/pom.xml +++ b/sdk/pom.xml @@ -33,12 +33,15 @@ UTF-8 - 1.5.4.RELEASE + 2.6.3 + 1.3.8.RELEASE 2.12.3 2.17.1 4.5.13 4.4.9 + 1.9.0 4.13.1 + 4.9.3 @@ -155,10 +158,38 @@ + org.springframework.boot spring-boot-starter-web ${spring-version} + + + org.springframework.boot + spring-boot-starter-logging + + + + ch.qos.logback + logback-classic + + + + + org.springframework.boot + spring-boot-starter-log4j + ${spring-boot-starter-log4j-version} + + + log4j + log4j + + + + + org.springframework.boot + spring-boot-starter-test + ${spring-version} org.apache.httpcomponents @@ -200,19 +231,38 @@ org.apache.flume flume-ng-core - 1.9.0 + ${flume-version} + + + log4j + log4j + + org.apache.flume flume-ng-node - 1.9.0 + ${flume-version} + + + log4j + log4j + + + junit junit ${junit-version} test + + com.squareup.okhttp3 + mockwebserver + ${mockwebserver-version} + test + diff --git a/sdk/src/main/java/com/qiniu/pandora/DefaultPandoraClient.java b/sdk/src/main/java/com/qiniu/pandora/DefaultPandoraClient.java index 977fd1e..aa6f43b 100644 --- a/sdk/src/main/java/com/qiniu/pandora/DefaultPandoraClient.java +++ b/sdk/src/main/java/com/qiniu/pandora/DefaultPandoraClient.java @@ -42,6 +42,10 @@ public PostDataService NewPostDataService() { return new PostDataService(this); } + public PostDataService NewPostDataService(TokenService tokenService, String token) { + return new PostDataService(this, tokenService, token); + } + public TokenService NewTokenService() { return new TokenService(this); } diff --git a/sdk/src/main/java/com/qiniu/pandora/collect/Collector.java b/sdk/src/main/java/com/qiniu/pandora/collect/Collector.java index 4066293..bc6fbd1 100644 --- a/sdk/src/main/java/com/qiniu/pandora/collect/Collector.java +++ b/sdk/src/main/java/com/qiniu/pandora/collect/Collector.java @@ -1,6 +1,6 @@ package com.qiniu.pandora.collect; -import com.qiniu.pandora.collect.runner.config.CollectorConfig; +import com.qiniu.pandora.collect.runner.config.RunnerConfig; import java.util.List; public interface Collector { @@ -9,9 +9,9 @@ public interface Collector { void stop(); - void addRunner(CollectorConfig config); + void addRunner(RunnerConfig config); - void addRunners(List configs); + void addRunners(List configs); void deleteRunner(String id); @@ -21,17 +21,17 @@ public interface Collector { void resetRunners(List id); - void updateRunner(CollectorConfig config); + void updateRunner(RunnerConfig config); - void updateRunners(List configs); + void updateRunners(List configs); void stopRunners(List ids); void startRunners(List ids); - List getAllRunners(); + List getAllRunners(); - List getRunners(List ids); + List getRunners(List ids); - CollectorConfig getRunner(String id); + RunnerConfig getRunner(String id); } diff --git a/sdk/src/main/java/com/qiniu/pandora/collect/CollectorBuilder.java b/sdk/src/main/java/com/qiniu/pandora/collect/CollectorBuilder.java deleted file mode 100644 index 22ecc4c..0000000 --- a/sdk/src/main/java/com/qiniu/pandora/collect/CollectorBuilder.java +++ /dev/null @@ -1,9 +0,0 @@ -package com.qiniu.pandora.collect; - -public class CollectorBuilder { - - // todo add collector config - public static Collector build() { - return new DefaultCollector(); - } -} diff --git a/sdk/src/main/java/com/qiniu/pandora/collect/CollectorConfig.java b/sdk/src/main/java/com/qiniu/pandora/collect/CollectorConfig.java new file mode 100644 index 0000000..7ab81c4 --- /dev/null +++ b/sdk/src/main/java/com/qiniu/pandora/collect/CollectorConfig.java @@ -0,0 +1,25 @@ +package com.qiniu.pandora.collect; + +import org.springframework.util.ObjectUtils; + +public class CollectorConfig { + public static final String DEFAULT_META_PATH = "./meta"; + + private String metaPath; + + public CollectorConfig() { + metaPath = DEFAULT_META_PATH; + } + + public CollectorConfig(String metaPath) { + if (ObjectUtils.isEmpty(metaPath)) { + this.metaPath = DEFAULT_META_PATH; + } else { + this.metaPath = metaPath; + } + } + + public String getMetaPath() { + return metaPath; + } +} diff --git a/sdk/src/main/java/com/qiniu/pandora/collect/CollectorContext.java b/sdk/src/main/java/com/qiniu/pandora/collect/CollectorContext.java new file mode 100644 index 0000000..047b685 --- /dev/null +++ b/sdk/src/main/java/com/qiniu/pandora/collect/CollectorContext.java @@ -0,0 +1,24 @@ +package com.qiniu.pandora.collect; + +import com.qiniu.pandora.service.token.TokenService; +import com.qiniu.pandora.service.upload.PostDataService; + +public class CollectorContext { + private PostDataService postDataService; + private TokenService tokenService; + + public CollectorContext() {} + + public CollectorContext(TokenService tokenService, PostDataService postDataService) { + this.postDataService = postDataService; + this.tokenService = tokenService; + } + + public PostDataService getPostDataService() { + return postDataService; + } + + public TokenService getTokenService() { + return tokenService; + } +} diff --git a/sdk/src/main/java/com/qiniu/pandora/collect/DefaultCollector.java b/sdk/src/main/java/com/qiniu/pandora/collect/DefaultCollector.java index 7716843..0a31d51 100644 --- a/sdk/src/main/java/com/qiniu/pandora/collect/DefaultCollector.java +++ b/sdk/src/main/java/com/qiniu/pandora/collect/DefaultCollector.java @@ -2,11 +2,12 @@ import com.qiniu.pandora.collect.runner.CollectRunner; import com.qiniu.pandora.collect.runner.Runner; -import com.qiniu.pandora.collect.runner.config.CollectorConfig; +import com.qiniu.pandora.collect.runner.config.RunnerConfig; import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import org.apache.flume.lifecycle.LifecycleSupervisor; import org.apache.log4j.LogManager; import org.apache.log4j.Logger; @@ -14,11 +15,17 @@ public class DefaultCollector implements Collector { private static final Logger logger = LogManager.getLogger(DefaultCollector.class); - private State state; + private final LifecycleSupervisor supervisor; private Map runners = new ConcurrentHashMap<>(); + private State state; + private CollectorConfig collectorConfig; + private CollectorContext context; - public DefaultCollector() { + public DefaultCollector(CollectorConfig collectorConfig, CollectorContext context) { + this.collectorConfig = collectorConfig; + this.context = context; state = State.NEW; + supervisor = new LifecycleSupervisor(); } @Override @@ -27,6 +34,7 @@ public void start() { throw new IllegalStateException("Cannot be started while started"); } state = State.STARTED; + supervisor.start(); runners.values().forEach(Runner::start); } @@ -37,15 +45,18 @@ public void stop() { } state = State.STOPPED; runners.values().forEach(Runner::stop); + supervisor.stop(); } @Override - public void addRunner(CollectorConfig config) { + public void addRunner(RunnerConfig config) { if (runners.containsKey(config.getId())) { throw new IllegalArgumentException("already add task: " + config.getId()); } runners.putIfAbsent( - config.getId(), new CollectRunner(config.getName(), config.getProperties())); + config.getId(), + new CollectRunner( + config.getName(), config.getProperties(), null, collectorConfig, context, supervisor)); Runner runner = runners.get(config.getId()); if (config.getState() != null && config.getState().equals(State.STARTED.toString())) { runner.start(); @@ -53,7 +64,7 @@ public void addRunner(CollectorConfig config) { } @Override - public void addRunners(List configs) { + public void addRunners(List configs) { if (configs == null || configs.isEmpty()) { return; } @@ -96,13 +107,13 @@ public void resetRunners(List ids) { } @Override - public void updateRunner(CollectorConfig config) { + public void updateRunner(RunnerConfig config) { Runner runner = runners.get(config.getId()); runner.update(config.getProperties()); } @Override - public void updateRunners(List configs) { + public void updateRunners(List configs) { if (configs == null || configs.isEmpty()) { return; } @@ -138,28 +149,36 @@ public void startRunners(List ids) { } @Override - public List getAllRunners() { - List configs = new ArrayList<>(runners.size()); + public List getAllRunners() { + List configs = new ArrayList<>(runners.size()); runners.forEach( (id, runner) -> configs.add( - new CollectorConfig( - id, runner.getName(), runner.getState(), runner.getProperties()))); + new RunnerConfig( + id, + runner.getName(), + runner.getState(), + runner.getMetadata(), + runner.getProperties()))); return configs; } @Override - public List getRunners(List ids) { - List configs = new ArrayList<>(ids.size()); + public List getRunners(List ids) { + List configs = new ArrayList<>(ids.size()); runners.forEach( (id, runner) -> { if (ids.contains(id)) { configs.add( - new CollectorConfig( - id, runner.getName(), runner.getState(), runner.getProperties())); + new RunnerConfig( + id, + runner.getName(), + runner.getState(), + runner.getMetadata(), + runner.getProperties())); } }); @@ -167,11 +186,12 @@ public List getRunners(List ids) { } @Override - public CollectorConfig getRunner(String id) { + public RunnerConfig getRunner(String id) { Runner runner = runners.get(id); if (runner == null) { return null; } - return new CollectorConfig(id, runner.getName(), runner.getState(), runner.getProperties()); + return new RunnerConfig( + id, runner.getName(), runner.getState(), runner.getMetadata(), runner.getProperties()); } } diff --git a/sdk/src/main/java/com/qiniu/pandora/collect/runner/CollectRunner.java b/sdk/src/main/java/com/qiniu/pandora/collect/runner/CollectRunner.java index 17661d7..6076711 100644 --- a/sdk/src/main/java/com/qiniu/pandora/collect/runner/CollectRunner.java +++ b/sdk/src/main/java/com/qiniu/pandora/collect/runner/CollectRunner.java @@ -1,8 +1,11 @@ package com.qiniu.pandora.collect.runner; +import com.qiniu.pandora.collect.CollectorConfig; +import com.qiniu.pandora.collect.CollectorContext; import com.qiniu.pandora.collect.State; import com.qiniu.pandora.collect.runner.config.EmbeddedRunnerConfiguration; import com.qiniu.pandora.collect.runner.config.MaterializedConfigurationProvider; +import com.qiniu.pandora.collect.runner.source.MetaSource; import java.util.Map; import org.apache.flume.Channel; import org.apache.flume.FlumeException; @@ -13,49 +16,91 @@ import org.apache.flume.lifecycle.LifecycleSupervisor; import org.apache.flume.lifecycle.LifecycleSupervisor.SupervisorPolicy; import org.apache.flume.node.MaterializedConfiguration; -import org.apache.log4j.LogManager; -import org.apache.log4j.Logger; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +/** 采集执行器，负责管理source、channel、sink的生命周期 */ public class CollectRunner implements Runner { private static final Logger logger = LogManager.getLogger(CollectRunner.class); private final MaterializedConfigurationProvider configurationProvider; - private final LifecycleSupervisor supervisor; + private LifecycleSupervisor supervisor; private SourceRunner sourceRunner; private Channel channel; private SinkRunner sinkRunner; - private String name; private Map properties; - + private String name; private State state; + private CollectorConfig config; + private CollectorContext context; - public CollectRunner(String name, Map properties) { - this(name, properties, new MaterializedConfigurationProvider()); + public CollectRunner( + String name, + Map properties, + String metadata, + CollectorConfig config, + CollectorContext context, + LifecycleSupervisor supervisor) { + this( + name, + properties, + metadata, + config, + context, + supervisor, + new MaterializedConfigurationProvider()); } public CollectRunner( String name, Map properties, + String metadata, + CollectorConfig config, + CollectorContext context, + LifecycleSupervisor supervisor, MaterializedConfigurationProvider configurationProvider) { this.name = name; this.properties = properties; this.configurationProvider = configurationProvider; - supervisor = new LifecycleSupervisor(); this.state = State.STOPPED; + this.supervisor = supervisor; + this.config = config; + this.context = context; + configure(properties, metadata, config); + state = State.NEW; } + @Override public void start() { if (state == State.STARTED) { - throw new IllegalStateException("Cannot be started while started"); - } else if (state == State.NEW) { - throw new IllegalStateException("Cannot be started before being " + "configured"); + logger.info("Runner [{}] has already started", name); + return; + } + + boolean error = true; + try { + channel.start(); + sinkRunner.start(); + sourceRunner.start(); + + supervisor.supervise( + channel, new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START); + supervisor.supervise( + sinkRunner, new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START); + supervisor.supervise( + sourceRunner, new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START); + error = false; + } finally { + if (error) { + stopLogError(sourceRunner); + stopLogError(channel); + stopLogError(sinkRunner); + } } - configure(properties); - doStart(); state = State.STARTED; - logger.info("runner [" + name + "] has started"); + logger.info("Runner [{}] has started", name); } @Override @@ -63,35 +108,40 @@ public void update(Map properties) { if (state == State.STARTED) { stop(); } - this.properties = properties; + configure(properties, "", config); start(); - logger.info("runner [" + name + "] has updated"); + logger.info("Runner [{}] has updated", name); } @Override public void reset() { if (state == State.STARTED) { - stop(); + delete(); } - // todo delete offset,backup data start(); - logger.info("runner [" + name + "] has reset"); + logger.info("Runner [{}] has reset", name); } + @Override public void stop() { if (state != State.STARTED) { return; } - supervisor.stop(); + + supervisor.unsupervise(sourceRunner); + supervisor.unsupervise(channel); + supervisor.unsupervise(sinkRunner); state = State.STOPPED; - logger.info("runner [" + name + "] has stopped"); + logger.info("Runner [{}] has stopped", name); } @Override public void delete() { stop(); - // todo delete meta - logger.info("runner [" + name + "] has deleted"); + if (sourceRunner.getSource() instanceof MetaSource) { + ((MetaSource) sourceRunner.getSource()).deleteMeta(); + } + logger.info("Runner [{}] has deleted", name); } @Override @@ -109,10 +159,20 @@ public State getState() { return this.state; } - private void configure(Map properties) { + @Override + public String getMetadata() { + if (sourceRunner.getSource() instanceof MetaSource) { + return ((MetaSource) sourceRunner.getSource()).getMeta(); + } + return null; + } - properties = EmbeddedRunnerConfiguration.configure(name, properties); - MaterializedConfiguration conf = configurationProvider.get(name, properties); + private void configure(Map properties, String metadata, CollectorConfig config) { + // 调整 properties + properties = + EmbeddedRunnerConfiguration.configure(name, properties, metadata, config.getMetaPath()); + // 创建 source/channel/sinks + MaterializedConfiguration conf = configurationProvider.get(name, properties, context); Map sources = conf.getSourceRunners(); if (sources.size() != 1) { throw new FlumeException("Expected one source and got " + sources.size()); @@ -130,37 +190,13 @@ private void configure(Map properties) { this.sinkRunner = sinks.values().iterator().next(); } - private void doStart() { - boolean error = true; - try { - channel.start(); - sinkRunner.start(); - sourceRunner.start(); - - supervisor.supervise( - channel, new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START); - supervisor.supervise( - sinkRunner, new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START); - supervisor.supervise( - sourceRunner, new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START); - error = false; - } finally { - if (error) { - stopLogError(sourceRunner); - stopLogError(channel); - stopLogError(sinkRunner); - supervisor.stop(); - } - } - } - private void stopLogError(LifecycleAware lifeCycleAware) { try { if (LifecycleState.START.equals(lifeCycleAware.getLifecycleState())) { lifeCycleAware.stop(); } } catch (Exception e) { - logger.warn("Exception while stopping " + lifeCycleAware, e); + logger.error("Runner [{}] failed while stopping", name, e); } } } diff --git a/sdk/src/main/java/com/qiniu/pandora/collect/runner/Runner.java b/sdk/src/main/java/com/qiniu/pandora/collect/runner/Runner.java index 89431f5..acc1959 100644 --- a/sdk/src/main/java/com/qiniu/pandora/collect/runner/Runner.java +++ b/sdk/src/main/java/com/qiniu/pandora/collect/runner/Runner.java @@ -20,4 +20,6 @@ public interface Runner { String getName(); State getState(); + + String getMetadata(); } diff --git a/sdk/src/main/java/com/qiniu/pandora/collect/runner/config/CustomMaterializedConfiguration.java b/sdk/src/main/java/com/qiniu/pandora/collect/runner/config/CustomMaterializedConfiguration.java new file mode 100644 index 0000000..f684bbf --- /dev/null +++ b/sdk/src/main/java/com/qiniu/pandora/collect/runner/config/CustomMaterializedConfiguration.java @@ -0,0 +1,3 @@ +package com.qiniu.pandora.collect.runner.config; + +public interface CustomMaterializedConfiguration {} diff --git a/sdk/src/main/java/com/qiniu/pandora/collect/runner/config/EmbeddedRunnerConfiguration.java b/sdk/src/main/java/com/qiniu/pandora/collect/runner/config/EmbeddedRunnerConfiguration.java index 2911bb7..05a1c0e 100644 --- a/sdk/src/main/java/com/qiniu/pandora/collect/runner/config/EmbeddedRunnerConfiguration.java +++ b/sdk/src/main/java/com/qiniu/pandora/collect/runner/config/EmbeddedRunnerConfiguration.java @@ -34,6 +34,14 @@ public class EmbeddedRunnerConfiguration { /** Space delimited list of sink names: e.g. sink1 sink2 sink3 */ public static final String SINKS = "sinks"; + /** Meta content for source */ + public static final String METADATA = "metadata"; + + public static final String META_PATH = "meta_path"; + + public static final String SOURCE_METADATA = join(SOURCE, METADATA); + public static final String SOURCE_META_PATH = join(SOURCE, META_PATH); + public static final String SINKS_PREFIX = join(SINKS, ""); /** Source type, choices are `embedded' */ public static final String SOURCE_TYPE = join(SOURCE, TYPE); @@ -122,18 +130,19 @@ private static void validate(String name, Map properties) throws * @param properties - embedded agent configuration * @return configuration applicable to a flume agent */ - public static Map configure(String name, Map properties) + public static Map configure( + String name, Map properties, String metadata, String metaPath) throws FlumeException { validate(name, properties); // we are going to modify the properties as we parse the config properties = new HashMap(properties); + // set source meta + properties.put(SOURCE_METADATA, metadata); + properties.put(SOURCE_META_PATH, metaPath); adapterCustomSourceAndSink(properties); String sinkNames = properties.remove(SINKS); String strippedName = name.replaceAll("\\s+", ""); - String sourceName = "source-" + strippedName; - String channelName = "channel-" + strippedName; - String sinkGroupName = "sink-group-" + strippedName; /* * Now we are going to process the user supplied configuration @@ -149,24 +158,24 @@ public static Map configure(String name, Map pro * source at the channel. */ // point agent at source - result.put(join(name, BasicConfigurationConstants.CONFIG_SOURCES), sourceName); + result.put(join(name, BasicConfigurationConstants.CONFIG_SOURCES), strippedName); // point agent at channel - result.put(join(name, BasicConfigurationConstants.CONFIG_CHANNELS), channelName); + result.put(join(name, BasicConfigurationConstants.CONFIG_CHANNELS), strippedName); // point agent at sinks result.put(join(name, BasicConfigurationConstants.CONFIG_SINKS), sinkNames); // points the agent at the sinkgroup - result.put(join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS), sinkGroupName); + result.put(join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS), strippedName); // points the sinkgroup at the sinks result.put( - join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS, sinkGroupName, SINKS), sinkNames); + join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS, strippedName, SINKS), sinkNames); // points the source at the channel result.put( join( name, BasicConfigurationConstants.CONFIG_SOURCES, - sourceName, + strippedName, BasicConfigurationConstants.CONFIG_CHANNELS), - channelName); + strippedName); // Properties will be modified during iteration so we need a // copy of the keys. @@ -191,7 +200,7 @@ public static Map configure(String name, Map pro BasicConfigurationConstants.CONFIG_SINKS, sink, BasicConfigurationConstants.CONFIG_CHANNEL), - channelName); + strippedName); } /* * Third, process all remaining configuration items, prefixing them @@ -202,16 +211,16 @@ public static Map configure(String name, Map pro String value = properties.get(key); if (key.startsWith(SOURCE_PREFIX)) { // users use `source' but agent needs the actual source name - key = key.replaceFirst(SOURCE, sourceName); + key = key.replaceFirst(SOURCE, strippedName); result.put(join(name, BasicConfigurationConstants.CONFIG_SOURCES, key), value); } else if (key.startsWith(CHANNEL_PREFIX)) { // users use `channel' but agent needs the actual channel name - key = key.replaceFirst(CHANNEL, channelName); + key = key.replaceFirst(CHANNEL, strippedName); result.put(join(name, BasicConfigurationConstants.CONFIG_CHANNELS, key), value); } else if (key.startsWith(SINK_PROCESSOR_PREFIX)) { // agent.sinkgroups.sinkgroup.processor.* result.put( - join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS, sinkGroupName, key), value); + join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS, strippedName, key), value); } else { // XXX should we simply ignore this? throw new FlumeException("Unknown configuration " + key); @@ -274,7 +283,7 @@ private static void adapterCustomSink(Map properties) { } String[] sinks = properties.get(EmbeddedRunnerConfiguration.SINKS).split("\\s+"); for (String sink : sinks) { - String key = join(sink, SEPERATOR, TYPE); + String key = join(sink, TYPE); com.qiniu.pandora.collect.runner.sinks.SinkType sinkType = com.qiniu.pandora.collect.runner.sinks.SinkType.fromString(properties.get(key)); if (sinkType != null) { diff --git a/sdk/src/main/java/com/qiniu/pandora/collect/runner/config/MaterializedConfigurationProvider.java b/sdk/src/main/java/com/qiniu/pandora/collect/runner/config/MaterializedConfigurationProvider.java index fc17856..dc002d6 100644 --- a/sdk/src/main/java/com/qiniu/pandora/collect/runner/config/MaterializedConfigurationProvider.java +++ b/sdk/src/main/java/com/qiniu/pandora/collect/runner/config/MaterializedConfigurationProvider.java @@ -1,5 +1,6 @@ package com.qiniu.pandora.collect.runner.config; +import com.qiniu.pandora.collect.CollectorContext; import java.util.Map; import org.apache.flume.node.MaterializedConfiguration; @@ -10,7 +11,8 @@ */ public class MaterializedConfigurationProvider { - public MaterializedConfiguration get(String name, Map properties) { - return new MemoryConfigurationProvider(name, properties).getConfiguration(); + public MaterializedConfiguration get( + String name, Map properties, CollectorContext context) { + return new MemoryConfigurationProvider(name, properties, context).getConfiguration(); } } diff --git a/sdk/src/main/java/com/qiniu/pandora/collect/runner/config/MemoryConfigurationProvider.java b/sdk/src/main/java/com/qiniu/pandora/collect/runner/config/MemoryConfigurationProvider.java index 88fb6bd..74da941 100644 --- a/sdk/src/main/java/com/qiniu/pandora/collect/runner/config/MemoryConfigurationProvider.java +++ b/sdk/src/main/java/com/qiniu/pandora/collect/runner/config/MemoryConfigurationProvider.java @@ -1,23 +1,575 @@ package com.qiniu.pandora.collect.runner.config; +import com.google.common.base.Preconditions; +import com.google.common.collect.ArrayListMultimap; +import com.google.common.collect.ListMultimap; +import com.google.common.collect.Lists; +import com.google.common.collect.Maps; +import com.qiniu.pandora.collect.CollectorContext; +import com.qiniu.pandora.collect.runner.sinks.ContextConfigurable; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Enumeration; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; import java.util.Map; +import java.util.Map.Entry; +import java.util.Properties; +import java.util.Set; +import org.apache.flume.Channel; +import org.apache.flume.ChannelFactory; +import org.apache.flume.ChannelSelector; +import org.apache.flume.Context; +import org.apache.flume.FlumeException; +import org.apache.flume.Sink; +import org.apache.flume.SinkFactory; +import org.apache.flume.SinkProcessor; +import org.apache.flume.SinkRunner; +import org.apache.flume.Source; +import org.apache.flume.SourceFactory; +import org.apache.flume.SourceRunner; +import org.apache.flume.annotations.Disposable; +import org.apache.flume.channel.ChannelProcessor; +import org.apache.flume.channel.ChannelSelectorFactory; +import org.apache.flume.channel.DefaultChannelFactory; +import org.apache.flume.conf.BasicConfigurationConstants; +import org.apache.flume.conf.BatchSizeSupported; +import org.apache.flume.conf.ComponentConfiguration; +import org.apache.flume.conf.Configurables; import org.apache.flume.conf.FlumeConfiguration; +import org.apache.flume.conf.FlumeConfiguration.AgentConfiguration; +import org.apache.flume.conf.TransactionCapacitySupported; +import org.apache.flume.conf.channel.ChannelSelectorConfiguration; +import org.apache.flume.conf.sink.SinkConfiguration; +import org.apache.flume.conf.sink.SinkGroupConfiguration; +import org.apache.flume.conf.source.SourceConfiguration; import org.apache.flume.node.AbstractConfigurationProvider; +import org.apache.flume.node.MaterializedConfiguration; +import org.apache.flume.node.SimpleMaterializedConfiguration; +import org.apache.flume.sink.DefaultSinkFactory; +import org.apache.flume.sink.DefaultSinkProcessor; +import org.apache.flume.sink.SinkGroup; +import org.apache.flume.source.DefaultSourceFactory; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * MemoryConfigurationProvider is the simplest possible AbstractConfigurationProvider simply turning * a give properties file and agent name into a FlumeConfiguration object. */ class MemoryConfigurationProvider extends AbstractConfigurationProvider { + + private static final Logger LOGGER = LoggerFactory.getLogger(MemoryConfigurationProvider.class); + + private final String agentName; + private final SourceFactory sourceFactory; + private final SinkFactory sinkFactory; + private final ChannelFactory channelFactory; + private final CollectorContext context; + + private final Map, Map> channelCache; + private final Map properties; - MemoryConfigurationProvider(String name, Map properties) { + MemoryConfigurationProvider( + String name, Map properties, CollectorContext context) { super(name); + this.agentName = name; + this.sourceFactory = new DefaultSourceFactory(); + this.sinkFactory = new DefaultSinkFactory(); + this.channelFactory = new DefaultChannelFactory(); this.properties = properties; + this.context = context; + channelCache = new HashMap<>(); } @Override protected FlumeConfiguration getFlumeConfiguration() { return new FlumeConfiguration(properties); } + + public MaterializedConfiguration getConfiguration() { + MaterializedConfiguration conf = new SimpleMaterializedConfiguration(); + FlumeConfiguration fconfig = getFlumeConfiguration(); + AgentConfiguration agentConf = fconfig.getConfigurationFor(getAgentName()); + if (agentConf != null) { + Map channelComponentMap = Maps.newHashMap(); + Map sourceRunnerMap = Maps.newHashMap(); + Map sinkRunnerMap = Maps.newHashMap(); + try { + loadChannels(agentConf, channelComponentMap); + loadSources(agentConf, channelComponentMap, sourceRunnerMap); + loadSinks(agentConf, channelComponentMap, sinkRunnerMap); + Set channelNames = new HashSet(channelComponentMap.keySet()); + for (String channelName : channelNames) { + ChannelComponent channelComponent = channelComponentMap.get(channelName); + if (channelComponent.components.isEmpty()) { + LOGGER.warn( + String.format( + "Channel %s has no components connected" + " and has been removed.", + channelName)); + channelComponentMap.remove(channelName); + Map nameChannelMap = + channelCache.get(channelComponent.channel.getClass()); + if (nameChannelMap != null) { + nameChannelMap.remove(channelName); + } + } else { + LOGGER.info( + String.format( + "Channel %s connected to %s", + channelName, channelComponent.components.toString())); + conf.addChannel(channelName, channelComponent.channel); + } + } + for (Map.Entry entry : sourceRunnerMap.entrySet()) { + conf.addSourceRunner(entry.getKey(), entry.getValue()); + } + for (Map.Entry entry : sinkRunnerMap.entrySet()) { + conf.addSinkRunner(entry.getKey(), entry.getValue()); + } + } catch (InstantiationException ex) { + LOGGER.error("Failed to instantiate component", ex); + } finally { + channelComponentMap.clear(); + sourceRunnerMap.clear(); + sinkRunnerMap.clear(); + } + } else { + LOGGER.warn("No configuration found for this host:{}", getAgentName()); + } + return conf; + } + + public String getAgentName() { + return agentName; + } + + private void loadChannels( + AgentConfiguration agentConf, Map channelComponentMap) + throws InstantiationException { + LOGGER.info("Creating channels"); + + /* + * Some channels will be reused across re-configurations. To handle this, + * we store all the names of current channels, perform the reconfiguration, + * and then if a channel was not used, we delete our reference to it. + * This supports the scenario where you enable channel "ch0" then remove it + * and add it back. Without this, channels like memory channel would cause + * the first instances data to show up in the seconds. + */ + ListMultimap, String> channelsNotReused = ArrayListMultimap.create(); + // assume all channels will not be re-used + for (Map.Entry, Map> entry : + channelCache.entrySet()) { + Class channelKlass = entry.getKey(); + Set channelNames = entry.getValue().keySet(); + channelsNotReused.get(channelKlass).addAll(channelNames); + } + + Set channelNames = agentConf.getChannelSet(); + Map compMap = agentConf.getChannelConfigMap(); + /* + * Components which have a ComponentConfiguration object + */ + for (String chName : channelNames) { + ComponentConfiguration comp = compMap.get(chName); + if (comp != null) { + Channel channel = + getOrCreateChannel(channelsNotReused, comp.getComponentName(), comp.getType()); + try { + Configurables.configure(channel, comp); + channelComponentMap.put(comp.getComponentName(), new ChannelComponent(channel)); + LOGGER.info("Created channel " + chName); + } catch (Exception e) { + String msg = + String.format( + "Channel %s has been removed due to an " + "error during configuration", chName); + LOGGER.error(msg, e); + } + } + } + /* + * Components which DO NOT have a ComponentConfiguration object + * and use only Context + */ + for (String chName : channelNames) { + Context context = agentConf.getChannelContext().get(chName); + if (context != null) { + Channel channel = + getOrCreateChannel( + channelsNotReused, + chName, + context.getString(BasicConfigurationConstants.CONFIG_TYPE)); + try { + Configurables.configure(channel, context); + channelComponentMap.put(chName, new ChannelComponent(channel)); + LOGGER.info("Created channel " + chName); + } catch (Exception e) { + String msg = + String.format( + "Channel %s has been removed due to an " + "error during configuration", chName); + LOGGER.error(msg, e); + } + } + } + /* + * Any channel which was not re-used, will have it's reference removed + */ + for (Class channelKlass : channelsNotReused.keySet()) { + Map channelMap = channelCache.get(channelKlass); + if (channelMap != null) { + for (String channelName : channelsNotReused.get(channelKlass)) { + if (channelMap.remove(channelName) != null) { + LOGGER.info("Removed {} of type {}", channelName, channelKlass); + } + } + if (channelMap.isEmpty()) { + channelCache.remove(channelKlass); + } + } + } + } + + private Channel getOrCreateChannel( + ListMultimap, String> channelsNotReused, String name, String type) + throws FlumeException { + + Class channelClass = channelFactory.getClass(type); + /* + * Channel has requested a new instance on each re-configuration + */ + if (channelClass.isAnnotationPresent(Disposable.class)) { + Channel channel = channelFactory.create(name, type); + channel.setName(name); + return channel; + } + Map channelMap = channelCache.get(channelClass); + if (channelMap == null) { + channelMap = new HashMap(); + channelCache.put(channelClass, channelMap); + } + Channel channel = channelMap.get(name); + if (channel == null) { + channel = channelFactory.create(name, type); + channel.setName(name); + channelMap.put(name, channel); + } + channelsNotReused.get(channelClass).remove(name); + return channel; + } + + private void loadSources( + AgentConfiguration agentConf, + Map channelComponentMap, + Map sourceRunnerMap) + throws InstantiationException { + + Set sourceNames = agentConf.getSourceSet(); + Map compMap = agentConf.getSourceConfigMap(); + /* + * Components which have a ComponentConfiguration object + */ + for (String sourceName : sourceNames) { + ComponentConfiguration comp = compMap.get(sourceName); + if (comp != null) { + SourceConfiguration config = (SourceConfiguration) comp; + + Source source = sourceFactory.create(comp.getComponentName(), comp.getType()); + try { + Configurables.configure(source, config); + Set channelNames = config.getChannels(); + List sourceChannels = + getSourceChannels(channelComponentMap, source, channelNames); + if (sourceChannels.isEmpty()) { + String msg = String.format("Source %s is not connected to a " + "channel", sourceName); + throw new IllegalStateException(msg); + } + ChannelSelectorConfiguration selectorConfig = config.getSelectorConfiguration(); + + ChannelSelector selector = ChannelSelectorFactory.create(sourceChannels, selectorConfig); + + ChannelProcessor channelProcessor = new ChannelProcessor(selector); + Configurables.configure(channelProcessor, config); + + source.setChannelProcessor(channelProcessor); + sourceRunnerMap.put(comp.getComponentName(), SourceRunner.forSource(source)); + for (Channel channel : sourceChannels) { + ChannelComponent channelComponent = + Preconditions.checkNotNull( + channelComponentMap.get(channel.getName()), + String.format("Channel %s", channel.getName())); + channelComponent.components.add(sourceName); + } + } catch (Exception e) { + String msg = + String.format( + "Source %s has been removed due to an " + "error during configuration", + sourceName); + LOGGER.error(msg, e); + } + } + } + /* + * Components which DO NOT have a ComponentConfiguration object + * and use only Context + */ + Map sourceContexts = agentConf.getSourceContext(); + for (String sourceName : sourceNames) { + Context context = sourceContexts.get(sourceName); + if (context != null) { + Source source = + sourceFactory.create( + sourceName, context.getString(BasicConfigurationConstants.CONFIG_TYPE)); + try { + Configurables.configure(source, context); + String[] channelNames = + context.getString(BasicConfigurationConstants.CONFIG_CHANNELS).split("\\s+"); + List sourceChannels = + getSourceChannels(channelComponentMap, source, Arrays.asList(channelNames)); + if (sourceChannels.isEmpty()) { + String msg = String.format("Source %s is not connected to a " + "channel", sourceName); + throw new IllegalStateException(msg); + } + Map selectorConfig = + context.getSubProperties( + BasicConfigurationConstants.CONFIG_SOURCE_CHANNELSELECTOR_PREFIX); + + ChannelSelector selector = ChannelSelectorFactory.create(sourceChannels, selectorConfig); + + ChannelProcessor channelProcessor = new ChannelProcessor(selector); + Configurables.configure(channelProcessor, context); + source.setChannelProcessor(channelProcessor); + sourceRunnerMap.put(sourceName, SourceRunner.forSource(source)); + for (Channel channel : sourceChannels) { + ChannelComponent channelComponent = + Preconditions.checkNotNull( + channelComponentMap.get(channel.getName()), + String.format("Channel %s", channel.getName())); + channelComponent.components.add(sourceName); + } + } catch (Exception e) { + String msg = + String.format( + "Source %s has been removed due to an " + "error during configuration", + sourceName); + LOGGER.error(msg, e); + } + } + } + } + + private List getSourceChannels( + Map channelComponentMap, + Source source, + Collection channelNames) + throws InstantiationException { + List sourceChannels = new ArrayList(); + for (String chName : channelNames) { + ChannelComponent channelComponent = channelComponentMap.get(chName); + if (channelComponent != null) { + checkSourceChannelCompatibility(source, channelComponent.channel); + sourceChannels.add(channelComponent.channel); + } + } + return sourceChannels; + } + + private void checkSourceChannelCompatibility(Source source, Channel channel) + throws InstantiationException { + if (source instanceof BatchSizeSupported && channel instanceof TransactionCapacitySupported) { + long transCap = ((TransactionCapacitySupported) channel).getTransactionCapacity(); + long batchSize = ((BatchSizeSupported) source).getBatchSize(); + if (transCap < batchSize) { + String msg = + String.format( + "Incompatible source and channel settings defined. " + + "source's batch size is greater than the channels transaction capacity. " + + "Source: %s, batch size = %d, channel %s, transaction capacity = %d", + source.getName(), batchSize, channel.getName(), transCap); + throw new InstantiationException(msg); + } + } + } + + private void checkSinkChannelCompatibility(Sink sink, Channel channel) + throws InstantiationException { + if (sink instanceof BatchSizeSupported && channel instanceof TransactionCapacitySupported) { + long transCap = ((TransactionCapacitySupported) channel).getTransactionCapacity(); + long batchSize = ((BatchSizeSupported) sink).getBatchSize(); + if (transCap < batchSize) { + String msg = + String.format( + "Incompatible sink and channel settings defined. " + + "sink's batch size is greater than the channels transaction capacity. " + + "Sink: %s, batch size = %d, channel %s, transaction capacity = %d", + sink.getName(), batchSize, channel.getName(), transCap); + throw new InstantiationException(msg); + } + } + } + + private void loadSinks( + AgentConfiguration agentConf, + Map channelComponentMap, + Map sinkRunnerMap) + throws InstantiationException { + Set sinkNames = agentConf.getSinkSet(); + Map compMap = agentConf.getSinkConfigMap(); + Map sinks = new HashMap(); + /* + * Components which have a ComponentConfiguration object + */ + for (String sinkName : sinkNames) { + ComponentConfiguration comp = compMap.get(sinkName); + if (comp != null) { + SinkConfiguration config = (SinkConfiguration) comp; + Sink sink = sinkFactory.create(comp.getComponentName(), comp.getType()); + try { + if (sink instanceof ContextConfigurable) { + ((ContextConfigurable) sink).configContext(context); + } + Configurables.configure(sink, config); + ChannelComponent channelComponent = channelComponentMap.get(config.getChannel()); + if (channelComponent == null) { + String msg = String.format("Sink %s is not connected to a " + "channel", sinkName); + throw new IllegalStateException(msg); + } + checkSinkChannelCompatibility(sink, channelComponent.channel); + sink.setChannel(channelComponent.channel); + sinks.put(comp.getComponentName(), sink); + channelComponent.components.add(sinkName); + } catch (Exception e) { + String msg = + String.format( + "Sink %s has been removed due to an " + "error during configuration", sinkName); + LOGGER.error(msg, e); + } + } + } + /* + * Components which DO NOT have a ComponentConfiguration object + * and use only Context + */ + Map sinkContexts = agentConf.getSinkContext(); + for (String sinkName : sinkNames) { + Context context = sinkContexts.get(sinkName); + if (context != null) { + Sink sink = + sinkFactory.create( + sinkName, context.getString(BasicConfigurationConstants.CONFIG_TYPE)); + try { + if (sink instanceof ContextConfigurable) { + ((ContextConfigurable) sink).configContext(this.context); + } + Configurables.configure(sink, context); + ChannelComponent channelComponent = + channelComponentMap.get( + context.getString(BasicConfigurationConstants.CONFIG_CHANNEL)); + if (channelComponent == null) { + String msg = String.format("Sink %s is not connected to a " + "channel", sinkName); + throw new IllegalStateException(msg); + } + checkSinkChannelCompatibility(sink, channelComponent.channel); + sink.setChannel(channelComponent.channel); + sinks.put(sinkName, sink); + channelComponent.components.add(sinkName); + } catch (Exception e) { + String msg = + String.format( + "Sink %s has been removed due to an " + "error during configuration", sinkName); + LOGGER.error(msg, e); + } + } + } + + loadSinkGroups(agentConf, sinks, sinkRunnerMap); + } + + private void loadSinkGroups( + AgentConfiguration agentConf, Map sinks, Map sinkRunnerMap) + throws InstantiationException { + Set sinkGroupNames = agentConf.getSinkgroupSet(); + Map compMap = agentConf.getSinkGroupConfigMap(); + Map usedSinks = new HashMap(); + for (String groupName : sinkGroupNames) { + ComponentConfiguration comp = compMap.get(groupName); + if (comp != null) { + SinkGroupConfiguration groupConf = (SinkGroupConfiguration) comp; + List groupSinks = new ArrayList(); + for (String sink : groupConf.getSinks()) { + Sink s = sinks.remove(sink); + if (s == null) { + String sinkUser = usedSinks.get(sink); + if (sinkUser != null) { + throw new InstantiationException( + String.format( + "Sink %s of group %s already " + "in use by group %s", + sink, groupName, sinkUser)); + } else { + throw new InstantiationException( + String.format( + "Sink %s of group %s does " + "not exist or is not properly configured", + sink, groupName)); + } + } + groupSinks.add(s); + usedSinks.put(sink, groupName); + } + try { + SinkGroup group = new SinkGroup(groupSinks); + Configurables.configure(group, groupConf); + sinkRunnerMap.put(comp.getComponentName(), new SinkRunner(group.getProcessor())); + } catch (Exception e) { + String msg = + String.format( + "SinkGroup %s has been removed due to " + "an error during configuration", + groupName); + LOGGER.error(msg, e); + } + } + } + // add any unassigned sinks to solo collectors + for (Entry entry : sinks.entrySet()) { + if (!usedSinks.containsValue(entry.getKey())) { + try { + SinkProcessor pr = new DefaultSinkProcessor(); + List sinkMap = new ArrayList(); + sinkMap.add(entry.getValue()); + pr.setSinks(sinkMap); + Configurables.configure(pr, new Context()); + sinkRunnerMap.put(entry.getKey(), new SinkRunner(pr)); + } catch (Exception e) { + String msg = + String.format( + "SinkGroup %s has been removed due to " + "an error during configuration", + entry.getKey()); + LOGGER.error(msg, e); + } + } + } + } + + private static class ChannelComponent { + final Channel channel; + final List components; + + ChannelComponent(Channel channel) { + this.channel = channel; + components = Lists.newArrayList(); + } + } + + protected Map toMap(Properties properties) { + Map result = Maps.newHashMap(); + Enumeration propertyNames = properties.propertyNames(); + while (propertyNames.hasMoreElements()) { + String name = (String) propertyNames.nextElement(); + String value = properties.getProperty(name); + result.put(name, value); + } + return result; + } } diff --git a/sdk/src/main/java/com/qiniu/pandora/collect/runner/config/CollectorConfig.java b/sdk/src/main/java/com/qiniu/pandora/collect/runner/config/RunnerConfig.java similarity index 61% rename from sdk/src/main/java/com/qiniu/pandora/collect/runner/config/CollectorConfig.java rename to sdk/src/main/java/com/qiniu/pandora/collect/runner/config/RunnerConfig.java index 7d0eeaf..5b21a8b 100644 --- a/sdk/src/main/java/com/qiniu/pandora/collect/runner/config/CollectorConfig.java +++ b/sdk/src/main/java/com/qiniu/pandora/collect/runner/config/RunnerConfig.java @@ -4,26 +4,26 @@ import java.util.Map; import java.util.Objects; -public class CollectorConfig { +public class RunnerConfig { private String id; private String name; + private String metadata; private State state; private Map properties; - public CollectorConfig() {} + public RunnerConfig() {} - public CollectorConfig(String id, String name, Map properties) { - this.id = id; - this.name = name; - this.properties = properties; - this.state = State.NEW; + public RunnerConfig(String id, String name, String metadata, Map properties) { + this(id, name, State.NEW, metadata, properties); } - public CollectorConfig(String id, String name, State state, Map properties) { + public RunnerConfig( + String id, String name, State state, String metadata, Map properties) { this.id = id; this.name = name; this.state = state; + this.metadata = metadata; this.properties = properties; } @@ -39,19 +39,24 @@ public Map getProperties() { return properties; } + public String getMetadata() { + return metadata; + } + @Override public boolean equals(Object o) { if (this == o) return true; if (o == null || getClass() != o.getClass()) return false; - CollectorConfig that = (CollectorConfig) o; + RunnerConfig that = (RunnerConfig) o; return Objects.equals(id, that.id) && Objects.equals(name, that.name) - && Objects.equals(properties, that.properties); + && Objects.equals(properties, that.properties) + && Objects.equals(metadata, that.metadata); } @Override public int hashCode() { - return Objects.hash(id, name, properties); + return Objects.hash(id, name, properties, metadata); } public String getState() { diff --git a/sdk/src/main/java/com/qiniu/pandora/collect/runner/sinks/ContextConfigurable.java b/sdk/src/main/java/com/qiniu/pandora/collect/runner/sinks/ContextConfigurable.java new file mode 100644 index 0000000..6e5453b --- /dev/null +++ b/sdk/src/main/java/com/qiniu/pandora/collect/runner/sinks/ContextConfigurable.java @@ -0,0 +1,8 @@ +package com.qiniu.pandora.collect.runner.sinks; + +import com.qiniu.pandora.collect.CollectorContext; +import org.apache.flume.conf.Configurable; + +public interface ContextConfigurable extends Configurable { + void configContext(CollectorContext context); +} diff --git a/sdk/src/main/java/com/qiniu/pandora/collect/runner/sinks/PandoraSink.java b/sdk/src/main/java/com/qiniu/pandora/collect/runner/sinks/PandoraSink.java index 758a741..49824ce 100644 --- a/sdk/src/main/java/com/qiniu/pandora/collect/runner/sinks/PandoraSink.java +++ b/sdk/src/main/java/com/qiniu/pandora/collect/runner/sinks/PandoraSink.java @@ -1,44 +1,54 @@ package com.qiniu.pandora.collect.runner.sinks; -import com.qiniu.pandora.DefaultPandoraClient; -import com.qiniu.pandora.common.Constants; +import com.qiniu.pandora.collect.CollectorContext; import com.qiniu.pandora.service.upload.PostDataResponse; import com.qiniu.pandora.service.upload.PostDataService; +import com.qiniu.pandora.util.StringUtils; import java.io.IOException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; import org.apache.flume.Channel; import org.apache.flume.Context; import org.apache.flume.Event; import org.apache.flume.EventDeliveryException; import org.apache.flume.Transaction; -import org.apache.flume.conf.Configurable; import org.apache.flume.sink.AbstractSink; -import org.apache.log4j.LogManager; -import org.apache.log4j.Logger; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; import org.springframework.util.ObjectUtils; -public class PandoraSink extends AbstractSink implements Configurable { +public class PandoraSink extends AbstractSink implements ContextConfigurable { private static final Logger logger = LogManager.getLogger(PandoraSink.class); + // request + private static final String RAW = "raw"; + + // config name public static final String HOSTNAME = "hostname"; public static final String REPO = "repo"; public static final String SOURCE_TYPE = "source_type"; - public static final String PANDORA_HOST = "pandora_host"; - public static final String TOKEN = "token"; + public static final String BATCH_SIZE = "batch_size"; private String hostname; private String repo; private String sourceType; - private String token; + private int batchSize; private PostDataService postDataService; + @Override + public void configContext(CollectorContext context) { + this.postDataService = context.getPostDataService(); + } + @Override public final void configure(final Context context) { String hostname = context.getString(HOSTNAME, ""); - if (ObjectUtils.isEmpty(hostname)) { - hostname = "localhost"; - } String repo = context.getString(REPO); if (ObjectUtils.isEmpty(repo)) { @@ -50,81 +60,79 @@ public final void configure(final Context context) { throw new IllegalArgumentException("source_type cannot be empty"); } - String pandoraHost = context.getString(PANDORA_HOST); - if (ObjectUtils.isEmpty(pandoraHost)) { - throw new IllegalArgumentException("pandora_host cannot be empty"); - } - - String token = context.getString(TOKEN); - if (ObjectUtils.isEmpty(pandoraHost)) { - throw new IllegalArgumentException("pandora_host cannot be empty"); - } + int batchSize = context.getInteger(BATCH_SIZE, 2000); this.hostname = hostname; this.repo = repo; this.sourceType = sourceType; - this.token = token; - postDataService = new PostDataService(new DefaultPandoraClient(pandoraHost)); + this.batchSize = batchSize; + + if (postDataService == null) { + throw new IllegalArgumentException("post data service must be set"); + } } @Override public final void start() { - logger.info("Starting HttpSink"); + super.start(); + logger.info("Runner [{}] starting http sink", getName()); } @Override public final void stop() { - logger.info("Stopping HttpSink"); + super.stop(); + logger.info("Runner [{}] stopping http sink", getName()); + // todo send channel all data } @Override public final Status process() throws EventDeliveryException { - Status status = Status.READY; + Status status; - Channel ch = getChannel(); - Transaction txn = ch.getTransaction(); + Channel channel = getChannel(); + Transaction txn = channel.getTransaction(); txn.begin(); try { - Event event = ch.take(); - - byte[] eventBody = null; - if (event != null) { - eventBody = event.getBody(); + List> data = new ArrayList<>(batchSize); + List events = new ArrayList<>(batchSize); + while (data.size() < batchSize) { + Event event = channel.take(); + if (event == null) { + break; + } + byte[] eventBody = event.getBody(); + if (eventBody != null && eventBody.length > 0) { + data.add(combinePandoraBody(event)); + events.add(event); + } } - if (eventBody != null && eventBody.length > 0) { - logger.debug("Sending request : " + new String(event.getBody())); - - // todo check success / fail + if (!ObjectUtils.isEmpty(data)) { PostDataResponse response; try { - response = - postDataService.upload( - eventBody, - hostname, - "", - repo, - sourceType, - token, - Constants.CONTENT_TYPE_APPLICATION_JSON); + response = postDataService.upload(data, hostname, "", repo, sourceType); } catch (IOException e) { txn.rollback(); status = Status.BACKOFF; - logger.error("upload data failed", e); + logger.error("Runner [{}] upload data failed", getName(), e); + return status; } + handleResponse(events, response, channel); + txn.commit(); + status = Status.READY; } else { txn.commit(); - status = Status.BACKOFF; + status = Status.READY; - logger.warn("Processed empty event"); + logger.warn("Runner [{}] processed empty event", getName()); } } catch (Throwable t) { txn.rollback(); status = Status.BACKOFF; - logger.error("Error sending HTTP request, retrying", t); + logger.error("Runner [{}] failed to send pandora sink request, retrying", getName(), t); // re-throw all Errors if (t instanceof Error) { @@ -136,4 +144,54 @@ public final Status process() throws EventDeliveryException { } return status; } + + private void handleResponse(List events, PostDataResponse response, Channel channel) { + if (events.size() != response.getTotal()) { + logger.error( + "Runner [{}] pandora sink expect send {} data, but got total response {}", + getName(), + events.size(), + response.getTotal()); + } + if (ObjectUtils.isEmpty(response.getDetails())) { + return; + } + logger.error( + "Runner [{}] pandora sink expect send {} data, but success data count is {}, failed data count is {}, will retry to send failed data", + getName(), + response.getTotal(), + response.getSuccess(), + response.getFailure()); + Set errorMsg = new HashSet<>(response.getDetails().size()); + response + .getDetails() + .forEach( + detail -> { + if (detail.getStatus() / 200 == 1) { + return; + } + for (int i = detail.getStartPos(); i < detail.getEndPos(); i++) { + channel.put(events.get(i)); + } + if (!ObjectUtils.isEmpty(detail.getMessage())) { + errorMsg.add(detail.getMessage()); + } + }); + if (!ObjectUtils.isEmpty(errorMsg)) { + logger.error( + "Runner [{}] pandora sink failed message is [{}]", getName(), String.join(",", errorMsg)); + } + } + + private static Map combinePandoraBody(Event event) { + Map data; + if (ObjectUtils.isEmpty(event.getHeaders())) { + data = new HashMap<>(); + } else { + data = new HashMap<>(event.getHeaders().size() + 1); + data.putAll(event.getHeaders()); + } + data.put(RAW, StringUtils.utf8String(event.getBody())); + return data; + } } diff --git a/sdk/src/main/java/com/qiniu/pandora/collect/runner/source/FileSource.java b/sdk/src/main/java/com/qiniu/pandora/collect/runner/source/FileSource.java index d5ab2b1..153311a 100644 --- a/sdk/src/main/java/com/qiniu/pandora/collect/runner/source/FileSource.java +++ b/sdk/src/main/java/com/qiniu/pandora/collect/runner/source/FileSource.java @@ -1,14 +1,11 @@ package com.qiniu.pandora.collect.runner.source; -import java.io.BufferedReader; -import java.io.IOException; -import java.io.InputStream; -import java.io.InputStreamReader; -import java.nio.file.Files; -import java.nio.file.Path; -import java.nio.file.Paths; +import com.qiniu.pandora.collect.runner.config.EmbeddedRunnerConfiguration; +import com.qiniu.pandora.service.upload.PostDataService; +import java.io.RandomAccessFile; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.atomic.AtomicLong; import org.apache.flume.ChannelException; import org.apache.flume.Context; import org.apache.flume.Event; @@ -16,17 +13,26 @@ import org.apache.flume.conf.Configurable; import org.apache.flume.event.SimpleEvent; import org.apache.flume.source.AbstractSource; -import org.apache.log4j.LogManager; -import org.apache.log4j.Logger; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; import org.springframework.util.ObjectUtils; -public class FileSource extends AbstractSource implements EventDrivenSource, Configurable { +public class FileSource extends AbstractSource + implements EventDrivenSource, Configurable, MetaSource { private static final Logger logger = LogManager.getLogger(FileSource.class); public static final String INPUT_FILE = "input.file"; private String inputFile; + private String metaPath; + + private AtomicLong offset; + + public FileSource() { + super(); + offset = new AtomicLong(); + } @Override public void configure(Context context) { @@ -34,6 +40,14 @@ public void configure(Context context) { if (ObjectUtils.isEmpty(inputFile)) { throw new IllegalArgumentException("input.file cannot be empty"); } + metaPath = context.getString(EmbeddedRunnerConfiguration.META_PATH); + String meta = context.getString(EmbeddedRunnerConfiguration.METADATA); + if (ObjectUtils.isEmpty(meta)) { + String metadata = MetadataProcessor.read(getName(), metaPath); + offset.set(ObjectUtils.isEmpty(metadata) ? 0 : Long.parseLong(metadata)); + } else { + offset.set(Long.parseLong(meta)); + } } @Override @@ -43,26 +57,21 @@ public synchronized void stop() { @Override public void start() { - try { - Event event = new SimpleEvent(); - - Path file = Paths.get(inputFile); - try (InputStream in = Files.newInputStream(file); - BufferedReader reader = new BufferedReader(new InputStreamReader(in))) { - String line; - while ((line = reader.readLine()) != null) { - processLine(line.getBytes()); + super.start(); + try (RandomAccessFile raf = new RandomAccessFile(inputFile, "r")) { + while (raf.length() > offset.get()) { + raf.seek(offset.get()); + String line = raf.readLine(); + if (ObjectUtils.isEmpty(line)) { + continue; } - } catch (IOException e) { - logger.error("ERROR: ", e); + processLine(line.getBytes()); + offset.addAndGet(line.length()); + MetadataProcessor.save(String.valueOf(offset), getName(), metaPath); } - - // Store the Event into this Source's associated Channel(s) - getChannelProcessor().processEvent(event); - - } catch (Throwable t) { + } catch (Exception e) { // Log exception, handle individual exceptions as needed - logger.error("ERROR: ", t); + logger.error("Runner [{}] file source read failed", getName(), e); } } @@ -70,13 +79,25 @@ private void processLine(byte[] line) { byte[] message = line; Event event = new SimpleEvent(); Map headers = new HashMap<>(); + headers.put(PostDataService.ORIGIN, inputFile); headers.put("timestamp", String.valueOf(System.currentTimeMillis())); event.setBody(message); event.setHeaders(headers); try { + // Store the Event into this Source's associated Channel(s) getChannelProcessor().processEvent(event); } catch (ChannelException e) { - logger.error("ERROR: ", e); + logger.error("Runner [{}] FileSource process new line failed", getName(), e); } } + + @Override + public String getMeta() { + return String.valueOf(offset); + } + + @Override + public void deleteMeta() { + MetadataProcessor.delete(getName(), metaPath); + } } diff --git a/sdk/src/main/java/com/qiniu/pandora/collect/runner/source/MetaSource.java b/sdk/src/main/java/com/qiniu/pandora/collect/runner/source/MetaSource.java new file mode 100644 index 0000000..7734ed0 --- /dev/null +++ b/sdk/src/main/java/com/qiniu/pandora/collect/runner/source/MetaSource.java @@ -0,0 +1,8 @@ +package com.qiniu.pandora.collect.runner.source; + +public interface MetaSource { + + String getMeta(); + + void deleteMeta(); +} diff --git a/sdk/src/main/java/com/qiniu/pandora/collect/runner/source/MetadataProcessor.java b/sdk/src/main/java/com/qiniu/pandora/collect/runner/source/MetadataProcessor.java new file mode 100644 index 0000000..3c664ee --- /dev/null +++ b/sdk/src/main/java/com/qiniu/pandora/collect/runner/source/MetadataProcessor.java @@ -0,0 +1,57 @@ +package com.qiniu.pandora.collect.runner.source; + +import com.qiniu.pandora.util.StringUtils; +import java.io.File; +import java.io.FileWriter; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import org.apache.commons.io.FileUtils; +import org.apache.flume.FlumeException; + +public class MetadataProcessor { + + public static void save(String metadata, String runner, String metaDir) { + Path metaPath = Paths.get(metaDir); + try { + Files.createDirectories(metaPath); + } catch (IOException e) { + throw new FlumeException("Error creating positionFile parent directories", e); + } + + File file = new File(pathname(runner, metaDir)); + try (FileWriter writer = new FileWriter(file, false)) { + writer.write(metadata); + } catch (IOException e) { + throw new FlumeException("save metadata into local file failed", e); + } + } + + public static String read(String runner, String metaDir) throws FlumeException { + File file = new File(pathname(runner, metaDir)); + if (!file.exists()) { + return null; + } + try { + return StringUtils.utf8String(FileUtils.readFileToByteArray(file)); + } catch (IOException e) { + throw new FlumeException("read metadata from local file failed", e); + } + } + + public static void delete(String runner, String metaDir) { + File file = new File(pathname(runner, metaDir)); + try { + if (file.exists()) { + file.delete(); + } + } catch (Exception e) { + throw new FlumeException("delete local file failed", e); + } + } + + private static String pathname(String runner, String metaDir) { + return Paths.get(metaDir, String.format("%s.meta", runner)).toString(); + } +} diff --git a/sdk/src/main/java/com/qiniu/pandora/service/upload/PostDataResponse.java b/sdk/src/main/java/com/qiniu/pandora/service/upload/PostDataResponse.java index 4d5b094..c5cb037 100644 --- a/sdk/src/main/java/com/qiniu/pandora/service/upload/PostDataResponse.java +++ b/sdk/src/main/java/com/qiniu/pandora/service/upload/PostDataResponse.java @@ -1,17 +1,16 @@ package com.qiniu.pandora.service.upload; import java.util.List; -import java.util.Map; public class PostDataResponse { private int total; private int success; private int failure; - private List> details; + private List details; public PostDataResponse() {} - public PostDataResponse(int total, int success, int failure, List> details) { + public PostDataResponse(int total, int success, int failure, List details) { this.total = total; this.success = success; this.failure = failure; @@ -30,7 +29,39 @@ public int getFailure() { return failure; } - public List> getDetails() { + public List getDetails() { return details; } + + public static class ResponseItem { + private int startPos; + private int endPos; + private int status; + private String message; + + public ResponseItem() {} + + public ResponseItem(int startPos, int endPos, int status, String message) { + this.startPos = startPos; + this.endPos = endPos; + this.status = status; + this.message = message; + } + + public int getStartPos() { + return startPos; + } + + public int getEndPos() { + return endPos; + } + + public int getStatus() { + return status; + } + + public String getMessage() { + return message; + } + } } diff --git a/sdk/src/main/java/com/qiniu/pandora/service/upload/PostDataService.java b/sdk/src/main/java/com/qiniu/pandora/service/upload/PostDataService.java index fed3a8b..233080d 100644 --- a/sdk/src/main/java/com/qiniu/pandora/service/upload/PostDataService.java +++ b/sdk/src/main/java/com/qiniu/pandora/service/upload/PostDataService.java @@ -4,44 +4,102 @@ import com.qiniu.pandora.common.Constants; import com.qiniu.pandora.common.QiniuException; import com.qiniu.pandora.service.PandoraService; +import com.qiniu.pandora.service.token.TokenService; import com.qiniu.pandora.util.JsonHelper; +import com.qiniu.pandora.util.NetUtils; +import com.qiniu.pandora.util.StringUtils; import java.util.HashMap; +import java.util.List; import java.util.Map; +import org.springframework.util.ObjectUtils; public class PostDataService extends PandoraService { + private static final PostDataResponse emptyDataResponse = new PostDataResponse(0, 0, 0, null); + + public static final String ORIGIN = "origin"; private static final String HOST = "host"; - private static final String ORIGIN = "origin"; private static final String REPO = "repo"; private static final String SOURCE_TYPE = "sourcetype"; + private UploadTokenHolder tokenHolder; + private String defaultHostname; + public PostDataService(PandoraClient client) { + this(client, null, null); + } + + public PostDataService(PandoraClient client, TokenService tokenService, String token) { super(client); + defaultHostname = NetUtils.getHostname(); + this.tokenHolder = new UploadTokenHolder(tokenService, token); } public PostDataResponse upload( - byte[] rawBytes, + List> data, String host, String origin, String repo, String sourceType, - String token, - String contentType) + String token) throws QiniuException { + if (ObjectUtils.isEmpty(data)) { + return emptyDataResponse; + } + if (ObjectUtils.isEmpty(token)) { + throw new QiniuException("upload data token must be set"); + } Map params = new HashMap<>(); Map headers = new HashMap<>(); - params.put(HOST, host); + params.put(HOST, StringUtils.getStringOrDefault(host, defaultHostname)); params.put(ORIGIN, origin); params.put(REPO, repo); params.put(SOURCE_TYPE, sourceType); headers.put(Constants.AUTHORIZATION, token); byte[] body = client.post( - Constants.DEFAULT_DATA_PREFIX + combineParams(params), rawBytes, headers, contentType); + Constants.DEFAULT_DATA_PREFIX + combineParams(params), + JsonHelper.writeValueAsBytes(data), + headers, + Constants.CONTENT_TYPE_APPLICATION_JSON); PostDataResponse response = JsonHelper.readValue(PostDataResponse.class, body); if (response == null) { throw new QiniuException("parse upload data response failed"); } return response; } + + public PostDataResponse upload( + List> data, String host, String origin, String repo, String sourceType) + throws QiniuException { + return upload(data, host, origin, repo, sourceType, tokenHolder.getToken()); + } + + private static class UploadTokenHolder { + private TokenService tokenService; + + private static final long refreshInterval = 6 * 60 * 60 * 1000L; + private String token; + private long lastRefreshTime; + + public UploadTokenHolder(TokenService tokenService, String token) { + this.tokenService = tokenService; + this.token = token; + this.lastRefreshTime = 0L; + } + + public String getToken() throws QiniuException { + if (tokenService == null || ObjectUtils.isEmpty(token)) { + return null; + } + + synchronized (this) { + if (System.currentTimeMillis() - refreshInterval >= lastRefreshTime) { + token = tokenService.refreshToken(token); + lastRefreshTime = System.currentTimeMillis(); + } + } + return token; + } + } } diff --git a/sdk/src/main/java/com/qiniu/pandora/util/NetUtils.java b/sdk/src/main/java/com/qiniu/pandora/util/NetUtils.java index 86aaa09..b10bcc1 100644 --- a/sdk/src/main/java/com/qiniu/pandora/util/NetUtils.java +++ b/sdk/src/main/java/com/qiniu/pandora/util/NetUtils.java @@ -1,7 +1,9 @@ package com.qiniu.pandora.util; import java.io.IOException; +import java.net.InetAddress; import java.net.ServerSocket; +import java.net.UnknownHostException; public class NetUtils { @@ -9,4 +11,12 @@ public static int findFreePort() throws IOException { ServerSocket socket = new ServerSocket(0); return socket.getLocalPort(); } + + public static String getHostname() { + try { + return InetAddress.getLocalHost().getHostName(); + } catch (UnknownHostException ex) { + return "localhost"; + } + } } diff --git a/sdk/src/main/java/com/qiniu/pandora/util/StringUtils.java b/sdk/src/main/java/com/qiniu/pandora/util/StringUtils.java index d7dbc96..2b6f501 100644 --- a/sdk/src/main/java/com/qiniu/pandora/util/StringUtils.java +++ b/sdk/src/main/java/com/qiniu/pandora/util/StringUtils.java @@ -2,6 +2,7 @@ import com.qiniu.pandora.common.Constants; import java.util.Collection; +import org.springframework.util.ObjectUtils; /** 字符串连接工具类 */ public final class StringUtils { @@ -116,4 +117,8 @@ public static byte[] utf8Bytes(String data) { public static String utf8String(byte[] data) { return new String(data, Constants.UTF_8); } + + public static String getStringOrDefault(String value, String defaultValue) { + return !ObjectUtils.isEmpty(value) ? value : defaultValue; + } } diff --git a/sdk/src/test/java/com/qiniu/pandora/collect/CollectTests.java b/sdk/src/test/java/com/qiniu/pandora/collect/CollectTests.java index 1a54abf..df0e214 100644 --- a/sdk/src/test/java/com/qiniu/pandora/collect/CollectTests.java +++ b/sdk/src/test/java/com/qiniu/pandora/collect/CollectTests.java @@ -1,7 +1,7 @@ package com.qiniu.pandora.collect; import com.google.common.collect.Maps; -import com.qiniu.pandora.collect.runner.config.CollectorConfig; +import com.qiniu.pandora.collect.runner.config.RunnerConfig; import org.apache.flume.FlumeException; import org.junit.After; import org.junit.Before; @@ -13,7 +13,7 @@ public class CollectTests { @Before public void setUp() throws Exception { - collector = CollectorBuilder.build(); + collector = new DefaultCollector(new CollectorConfig(), new CollectorContext(null, null)); } @After @@ -27,7 +27,7 @@ public void tearDown() throws Exception { public void collectTests() throws Exception { collector.start(); try { - collector.addRunner(new CollectorConfig("1", "test", Maps.newHashMap())); + collector.addRunner(new RunnerConfig("1", "test", "", Maps.newHashMap())); } catch (FlumeException e) { } } diff --git a/sdk/src/test/java/com/qiniu/pandora/collect/runner/CollectRunnerTests.java b/sdk/src/test/java/com/qiniu/pandora/collect/runner/CollectRunnerTests.java index c00b418..445d0b6 100644 --- a/sdk/src/test/java/com/qiniu/pandora/collect/runner/CollectRunnerTests.java +++ b/sdk/src/test/java/com/qiniu/pandora/collect/runner/CollectRunnerTests.java @@ -1,24 +1,20 @@ package com.qiniu.pandora.collect.runner; -import com.google.common.base.Charsets; -import com.google.common.base.Preconditions; +import com.fasterxml.jackson.core.type.TypeReference; import com.google.common.collect.Maps; +import com.qiniu.pandora.DefaultPandoraClient; +import com.qiniu.pandora.collect.CollectorConfig; +import com.qiniu.pandora.collect.CollectorContext; +import com.qiniu.pandora.service.token.TokenService; +import com.qiniu.pandora.service.upload.PostDataService; +import com.qiniu.pandora.util.JsonHelper; import com.qiniu.pandora.util.NetUtils; -import java.net.InetSocketAddress; -import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.Queue; -import java.util.concurrent.LinkedBlockingQueue; -import org.apache.avro.AvroRemoteException; -import org.apache.avro.ipc.NettyServer; -import org.apache.avro.ipc.Responder; -import org.apache.avro.ipc.specific.SpecificResponder; -import org.apache.flume.Event; -import org.apache.flume.event.EventBuilder; -import org.apache.flume.source.avro.AvroFlumeEvent; -import org.apache.flume.source.avro.AvroSourceProtocol; -import org.apache.flume.source.avro.Status; +import okhttp3.mockwebserver.MockResponse; +import okhttp3.mockwebserver.MockWebServer; +import okhttp3.mockwebserver.RecordedRequest; +import org.apache.flume.lifecycle.LifecycleSupervisor; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -28,20 +24,17 @@ public class CollectRunnerTests { private static final String HOSTNAME = "localhost"; private CollectRunner runner; private Map properties; - private EventCollector eventCollector; - private NettyServer nettyServer; - private byte[] body; + private LifecycleSupervisor supervisor; + private CollectorConfig collectorConfig; + private CollectorContext context; + private MockWebServer server; @Before public void setUp() throws Exception { - body = "this is a test log".getBytes(Charsets.UTF_8); - int port = NetUtils.findFreePort(); - eventCollector = new EventCollector(); - Responder responderSink = new SpecificResponder(AvroSourceProtocol.class, eventCollector); - nettyServer = new NettyServer(responderSink, new InetSocketAddress(HOSTNAME, port)); - nettyServer.start(); - + supervisor = new LifecycleSupervisor(); + server = new MockWebServer(); + server.start(port); // give the server a second to start Thread.sleep(1000L); @@ -51,82 +44,104 @@ public void setUp() throws Exception { properties.put("channel.type", "memory"); properties.put("channel.capacity", "200"); properties.put("sinks", "sink1"); - properties.put("sink1.type", "avro"); + properties.put("sink1.type", "pandora"); properties.put("sink1.hostname", HOSTNAME); - properties.put("sink1.port", String.valueOf(port)); + properties.put("sink1.source_type", "json"); + properties.put("sink1.repo", "test"); properties.put("processor.type", "default"); - runner = new CollectRunner("test", properties); + collectorConfig = new CollectorConfig(this.getClass().getResource("/").getPath()); + DefaultPandoraClient client = new DefaultPandoraClient(String.format("127.0.0.1:%d", port)); + TokenService tokenService = client.NewTokenService(); + context = + new CollectorContext( + tokenService, new PostDataService(client, tokenService, "this is a fake token")); + runner = new CollectRunner("test", properties, null, collectorConfig, context, supervisor); } @After public void tearDown() throws Exception { if (runner != null) { - runner.stop(); + runner.delete(); + } + if (server != null) { + server.shutdown(); } - if (nettyServer != null) { - nettyServer.close(); + if (supervisor != null) { + supervisor.stop(); } } @Test(timeout = 30000L) public void testPut() throws Exception { runner.start(); - Event event; - while ((event = eventCollector.poll()) == null) { - Thread.sleep(500L); - } - Assert.assertNotNull(event); - Assert.assertArrayEquals(body, event.getBody()); + + // refresh token response + server.enqueue(new MockResponse().setResponseCode(200).setBody("{\"token\":\"new token\"}")); + RecordedRequest tokenRequest = server.takeRequest(); + + // upload data response + server.enqueue(new MockResponse().setResponseCode(200).setBody("{}")); + RecordedRequest request = server.takeRequest(); + Assert.assertNotNull(request); + List> body = + JsonHelper.readValue( + new TypeReference>>() {}, request.getBody().readByteArray()); + Assert.assertNotNull(body); + Assert.assertEquals(1, body.size()); + Assert.assertEquals("this is a test log", body.get(0).get("raw")); + Assert.assertEquals( + this.getClass().getResource("/test.log").getPath(), body.get(0).get("origin")); + Assert.assertNotNull(request.getRequestUrl()); + Assert.assertEquals("json", request.getRequestUrl().queryParameter("sourcetype")); + Assert.assertEquals("localhost", request.getRequestUrl().queryParameter("host")); + Assert.assertEquals("test", request.getRequestUrl().queryParameter("repo")); } @Test(timeout = 30000L) - public void testPutWithInterceptors() throws Exception { - properties.put("source.interceptors", "i1"); - properties.put("source.interceptors.i1.type", "static"); - properties.put("source.interceptors.i1.key", "key2"); - properties.put("source.interceptors.i1.value", "value2"); - + public void testPutWithMeta() throws Exception { + runner = new CollectRunner("test", properties, "5", collectorConfig, context, supervisor); runner.start(); - Event event; - while ((event = eventCollector.poll()) == null) { - Thread.sleep(500L); - } - Assert.assertNotNull(event); - Assert.assertArrayEquals(body, event.getBody()); + server.enqueue(new MockResponse().setResponseCode(200).setBody("{\"token\":\"new token\"}")); + RecordedRequest tokenRequest = server.takeRequest(); + + server.enqueue(new MockResponse().setResponseCode(200).setBody("{}")); + RecordedRequest request = server.takeRequest(); + Assert.assertNotNull(request); + List> body = + JsonHelper.readValue( + new TypeReference>>() {}, request.getBody().readByteArray()); + Assert.assertNotNull(body); + Assert.assertEquals(1, body.size()); + Assert.assertEquals("is a test log", body.get(0).get("raw")); } - static class EventCollector implements AvroSourceProtocol { - private final Queue eventQueue = new LinkedBlockingQueue(); - - public Event poll() { - AvroFlumeEvent avroEvent = eventQueue.poll(); - if (avroEvent != null) { - return EventBuilder.withBody( - avroEvent.getBody().array(), toStringMap(avroEvent.getHeaders())); - } - return null; - } - - @Override - public Status appendBatch(List events) throws AvroRemoteException { - Preconditions.checkState(eventQueue.addAll(events)); - return Status.OK; - } - - @Override - public Status append(AvroFlumeEvent avroFlumeEvent) throws AvroRemoteException { - eventQueue.add(avroFlumeEvent); - return Status.OK; - } - } + @Test(timeout = 30000L) + public void testPutWithFailed() throws Exception { + runner.start(); - private static Map toStringMap(Map charSeqMap) { - Map stringMap = new HashMap(); - for (Map.Entry entry : charSeqMap.entrySet()) { - stringMap.put(entry.getKey().toString(), entry.getValue().toString()); - } - return stringMap; + server.enqueue(new MockResponse().setResponseCode(200).setBody("{\"token\":\"new token\"}")); + RecordedRequest request = server.takeRequest(); + + server.enqueue( + new MockResponse() + .setResponseCode(200) + .setBody( + "{\"total\":1,\"success\":0,\"failure\":1,\"details\":[{\"startPos\":0,\"endPos\":1,\"status\":401,\"message\":\"test\"}]}")); + server.enqueue( + new MockResponse() + .setResponseCode(200) + .setBody("{\"total\":1,\"success\":1,\"failure\":0}")); + RecordedRequest firstRequest = server.takeRequest(); + Assert.assertNotNull(firstRequest); + RecordedRequest secondRequest = server.takeRequest(); + Assert.assertNotNull(secondRequest); + List> body = + JsonHelper.readValue( + new TypeReference>>() {}, + secondRequest.getBody().readByteArray()); + Assert.assertNotNull(body); + Assert.assertEquals(1, body.size()); } } diff --git a/sdk/src/test/resources/log4j2.properties b/sdk/src/test/resources/log4j2.properties new file mode 100644 index 0000000..404394a --- /dev/null +++ b/sdk/src/test/resources/log4j2.properties @@ -0,0 +1,7 @@ +appender.console.type = Console +appender.console.name = console +appender.console.layout.type = PatternLayout +appender.console.layout.pattern = [%d{ISO8601}][%-5p][%-25c{1.}] [%test_thread_info]%marker %m%n + +rootLogger.level = info +rootLogger.appenderRef.console.ref = console \ No newline at end of file