diff --git a/skywalking-analysis/pom.xml b/skywalking-analysis/pom.xml index 891f02163..064db08a7 100644 --- a/skywalking-analysis/pom.xml +++ b/skywalking-analysis/pom.xml @@ -25,16 +25,19 @@ org.apache.hbase hbase-client 1.1.2 + provided org.apache.hbase hbase-server 1.1.2 + provided org.apache.logging.log4j log4j-core 2.4.1 + provided com.ai.cloud @@ -77,6 +80,58 @@ ${project.build.sourceEncoding} + + org.apache.maven.plugins + maven-jar-plugin + + + + true + libs/ + com.ai.cloud.skywalking.analysis.AnalysisServerDriver + + + + + + package + + + + + + + + + + + + + + + + + + + + + maven-assembly-plugin + 2.4 + + + jar-with-dependencies + + + + + make-my-jar-with-dependencies + package + + single + + + + diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/AnalysisServerDriver.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/AnalysisServerDriver.java index 73d74476b..ff4d2fa4a 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/AnalysisServerDriver.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/AnalysisServerDriver.java @@ -1,25 +1,32 @@ package com.ai.cloud.skywalking.analysis; -import java.io.IOException; -import java.text.ParseException; -import java.text.SimpleDateFormat; -import java.util.Date; - +import com.ai.cloud.skywalking.analysis.categorize2chain.Categorize2ChainMapper; +import com.ai.cloud.skywalking.analysis.categorize2chain.Categorize2ChainReducer; +import com.ai.cloud.skywalking.analysis.categorize2chain.model.ChainInfo; +import com.ai.cloud.skywalking.analysis.config.Config; +import com.ai.cloud.skywalking.analysis.config.ConfigInitializer; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.MetaTableAccessor; +import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.io.ImmutableBytesWritable; import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil; +import org.apache.hadoop.io.RawComparator; +import org.apache.hadoop.mapred.JobContext; import org.apache.hadoop.mapreduce.Job; +import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.GenericOptionsParser; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.ai.cloud.skywalking.analysis.categorize2chain.Categorize2ChainMapper; -import com.ai.cloud.skywalking.analysis.categorize2chain.Categorize2ChainReduce; -import com.ai.cloud.skywalking.analysis.categorize2chain.model.ChainInfo; -import com.ai.cloud.skywalking.analysis.config.Config; +import java.io.IOException; +import java.text.ParseException; +import java.text.SimpleDateFormat; +import java.util.Date; public class AnalysisServerDriver extends Configured implements Tool { @@ -33,20 +40,29 @@ public class AnalysisServerDriver extends Configured implements Tool { @Override public int run(String[] args) throws Exception { + // ConfigInitializer.initialize(); Configuration conf = new Configuration(); + conf.set("hbase.zookeeper.quorum", Config.HBase.ZK_QUORUM); + conf.set("hbase.zookeeper.property.clientPort", Config.HBase.ZK_CLIENT_PORT); String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); if (otherArgs.length != 2) { System.err.println("Usage: AnalysisServer yyyy-MM-dd/HH:mm:ss yyyy-MM-dd/HH:mm:ss"); System.exit(2); } + Job job = Job.getInstance(conf); job.setJarByClass(AnalysisServerDriver.class); Scan scan = buildHBaseScan(args); - TableMapReduceUtil.initTableMapperJob(Config.HBase.CALL_CHAIN_TABLE_NAME, scan, Categorize2ChainMapper.class, - String.class, ChainInfo.class, job); - - job.setReducerClass(Categorize2ChainReduce.class); + TableMapReduceUtil.initTableMapperJob(Config.HBase.TABLE_CALL_CHAIN, scan, Categorize2ChainMapper.class, + ImmutableBytesWritable.class, ChainInfo.class, job); + int regions = MetaTableAccessor.getRegionCount(conf, TableName.valueOf(Config.HBase.TABLE_CHAIN_SUMMARY)); + if (regions == 0){ + regions = 1; + } + job.setReducerClass(Categorize2ChainReducer.class); + job.setNumReduceTasks(regions); + FileOutputFormat.setOutputPath(job, new Path("/tmp/mr/mySummaryFile")); return job.waitForCompletion(true) ? 0 : 1; } diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/Categorize2ChainMapper.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/Categorize2ChainMapper.java index a5c10ae6d..1a45201c2 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/Categorize2ChainMapper.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/Categorize2ChainMapper.java @@ -6,20 +6,20 @@ import com.ai.cloud.skywalking.analysis.categorize2chain.model.ChainInfo; import com.ai.cloud.skywalking.analysis.categorize2chain.model.ChainNode; import com.ai.cloud.skywalking.analysis.util.HBaseUtil; import com.ai.cloud.skywalking.protocol.Span; - import org.apache.hadoop.hbase.Cell; import org.apache.hadoop.hbase.client.Result; import org.apache.hadoop.hbase.io.ImmutableBytesWritable; import org.apache.hadoop.hbase.mapreduce.TableMapper; import org.apache.hadoop.hbase.util.Bytes; -import org.apache.hadoop.io.Text; +import org.apache.hadoop.io.RawComparator; +import org.apache.hadoop.mapred.JobContext; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; import java.util.*; -public class Categorize2ChainMapper extends TableMapper { +public class Categorize2ChainMapper extends TableMapper { private Logger logger = LoggerFactory.getLogger(Categorize2ChainMapper.class.getName()); @Override @@ -34,8 +34,8 @@ public class Categorize2ChainMapper extends TableMapper { } chainInfo = spanToChainInfo(key.toString(), spanList); - - context.write(new Text(chainInfo.getUserId() + ":" + chainInfo.getEntranceNodeToken()), chainInfo); + logger.info("Success convert span to chain info...." + chainInfo.getCID()); + context.write(new ImmutableBytesWritable((chainInfo.getUserId() + ":" + chainInfo.getEntranceNodeToken()).getBytes()), chainInfo); } catch (Exception e) { logger.error("Failed to mapper call chain[" + key.toString() + "]", e); } diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/Categorize2ChainReduce.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/Categorize2ChainReducer.java similarity index 64% rename from skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/Categorize2ChainReduce.java rename to skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/Categorize2ChainReducer.java index 66965d025..6dd77bc54 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/Categorize2ChainReduce.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/Categorize2ChainReducer.java @@ -1,26 +1,29 @@ package com.ai.cloud.skywalking.analysis.categorize2chain; +import com.ai.cloud.skywalking.analysis.categorize2chain.model.ChainInfo; +import com.ai.cloud.skywalking.analysis.config.ConfigInitializer; +import com.ai.cloud.skywalking.analysis.util.HBaseUtil; +import org.apache.hadoop.io.IntWritable; +import org.apache.hadoop.io.Text; +import org.apache.hadoop.mapreduce.Reducer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import java.io.IOException; import java.util.Iterator; -import org.apache.hadoop.hbase.client.Put; -import org.apache.hadoop.hbase.mapreduce.TableReducer; -import org.apache.hadoop.io.Text; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.ai.cloud.skywalking.analysis.categorize2chain.model.ChainInfo; -import com.ai.cloud.skywalking.analysis.util.HBaseUtil; - -public class Categorize2ChainReduce extends TableReducer { - private static Logger logger = LoggerFactory.getLogger(Categorize2ChainReduce.class.getName()); +public class Categorize2ChainReducer extends Reducer { + private static Logger logger = LoggerFactory.getLogger(Categorize2ChainReducer.class.getName()); @Override protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { - reduceAction(key.toString(), values.iterator()); + //ConfigInitializer.initialize(); + int totalCount = reduceAction(key.toString(), values.iterator()); + context.write(key, new IntWritable(totalCount)); } - public static void reduceAction(String key, Iterator chainInfoIterator) throws IOException, InterruptedException { + public static int reduceAction(String key, Iterator chainInfoIterator) throws IOException, InterruptedException { + int totalCount = 0; try { ChainRelate chainRelate = HBaseUtil.selectCallChainRelationship(key.toString()); Summary summary = new Summary(); @@ -32,6 +35,7 @@ public class Categorize2ChainReduce extends TableReducer { } catch (Exception e) { continue; } + totalCount++; } chainRelate.save(); @@ -39,5 +43,7 @@ public class Categorize2ChainReduce extends TableReducer { } catch (Exception e) { logger.error("Failed to reduce key[" + key + "]", e); } + + return totalCount; } } diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/dao/CallChainInfoDao.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/dao/CallChainInfoDao.java index 235a0583c..fcc79e49c 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/dao/CallChainInfoDao.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/dao/CallChainInfoDao.java @@ -17,10 +17,10 @@ public class CallChainInfoDao { static { try { - Class.forName(Config.MySql.driverClass); - connection = DriverManager.getConnection(Config.MySql.url, Config.MySql.userName, Config.MySql.password); + Class.forName(Config.MySql.DRIVER_CLASS); + connection = DriverManager.getConnection(Config.MySql.URL, Config.MySql.USERNAME, Config.MySql.PASSWORD); } catch (ClassNotFoundException e) { - logger.error("Failed to find jdbc driver class[" + Config.MySql.driverClass + "]", e); + logger.error("Failed to find jdbc driver class[" + Config.MySql.DRIVER_CLASS + "]", e); System.exit(-1); } catch (SQLException e) { logger.error("Failed to connection database.", e); diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/Config.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/Config.java index 7ab9d2087..2edad608e 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/Config.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/Config.java @@ -3,7 +3,6 @@ package com.ai.cloud.skywalking.analysis.config; public class Config { public static class HBase { - public static String TABLE_CHAIN_DETAIL = "sw_chain_detail"; public static String TRACE_DETAIL_FAMILY_COLUMN = "chain_detail"; @@ -11,22 +10,23 @@ public class Config { public static String TRACE_INFO_COLUMN_FAMILY = "trace_info"; - public static String CALL_CHAIN_TABLE_NAME = "sw-call-chain"; + public static String ZK_QUORUM = "10.1.235.197,10.1.235.198,10.1.235.199"; - public static String ZK_QUORUM; + public static String ZK_CLIENT_PORT = "29181"; - public static String ZK_CLIENT_PORT; + public static String TRACE_INFO_TABLE_NAME = "trace-info"; - public static String TRACE_INFO_TABLE_NAME = "trace_info"; + public static String TABLE_CALL_CHAIN_RELATIONSHIP = "sw-chain-relationship"; - public static String TABLE_CALL_CHAIN_RELATIONSHIP = "sw_chain_relationship"; + public static String CHAIN_RELATIONSHIP_COLUMN_FAMILY = "chain-relationship"; - public static String CHAIN_RELATIONSHIP_COLUMN_FAMILY = "chain_relationship"; + public static String TABLE_CHAIN_INFO = "sw-chain-info"; - public static String TABLE_CHAIN_INFO = "sw_chain_info"; + public static String TABLE_CHAIN_SUMMARY = "sw-chain-summary"; - public static String TABLE_CHAIN_SUMMARY = "sw_chain_summary"; + public static String TABLE_CHAIN_DETAIL = "sw-chain-detail"; + public static String TABLE_CALL_CHAIN = "sw-call-chain"; } public static class TraceInfo { @@ -35,17 +35,17 @@ public class Config { public static class MySql { - public static String url = "jdbc:mysql://10.1.228.202:31316/test"; + public static String URL = "jdbc:mysql://10.1.228.202:31316/test"; - public static String userName = "devrdbusr21"; + public static String USERNAME = "devrdbusr21"; - public static String password = "devrdbusr21"; + public static String PASSWORD = "devrdbusr21"; - public static String driverClass = "com.mysql.jdbc.Driver"; + public static String DRIVER_CLASS = "com.mysql.jdbc.Driver"; } public static class Filter { - public static String FILTER_PACKAGE_NAME; + public static String FILTER_PACKAGE_NAME = "com.ai.cloud.skywalking.analysis.categorize2chain.filter.impl"; } public class ChainNodeSummary { diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/ConfigInitializer.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/ConfigInitializer.java index bd3305b17..b67ad3130 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/ConfigInitializer.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/ConfigInitializer.java @@ -13,7 +13,9 @@ public class ConfigInitializer { private static Logger logger = Logger.getLogger(ConfigInitializer.class.getName()); public static void initialize() { - InputStream inputStream = ConfigInitializer.class.getResourceAsStream("/config.properties"); + InputStream inputStream = Thread.currentThread() + .getContextClassLoader().getResourceAsStream("/config.properties"); + //InputStream inputStream = ConfigInitializer.class.getResourceAsStream("/config.properties"); if (inputStream == null) { logger.log(Level.ALL, "No provider sky-walking certification documents, sky-walking api auto shutdown."); } else { diff --git a/skywalking-analysis/src/main/resources/config.properties b/skywalking-analysis/src/main/resources/config.properties index dcf06da68..050bf9b02 100644 --- a/skywalking-analysis/src/main/resources/config.properties +++ b/skywalking-analysis/src/main/resources/config.properties @@ -1,14 +1,35 @@ -#拦截器的类中的包名 -filter.filter_package_name=com.ai.cloud.skywalking.analysis.filter.impl -hbase.call_chain_table_name=sw-call-chain -hbase.zk_quorum=10.1.235.197,10.1.235.198,10.1.235.199 -hbase.zk_client_port=29181 -hbase.trace_info_table_name=sw-trace-info +hbase.trace_detail_family_column=chain_detail + +hbase.chain_summary_column_family=chain_summary + hbase.trace_info_column_family=trace_info +hbase.zk_quorum=10.1.235.197,10.1.235.198,10.1.235.199 + +hbase.zk_client_port=29181 + +hbase.trace_info_table_name=trace-info + +hbase.table_call_chain_relationship=sw-chain-relationship + +hbase.chain_relationship_column_family=chain-relationship + +hbase.table_chain_info=sw-chain-info + +hbase.table_chain_summary=sw-chain-summary + +hbase.table_chain_detail=sw-chain-detail + +hbase.table_call_chain=sw-call-chain + + traceinfo.trace_info_column_cid=cid mysql.url=jdbc:mysql://10.1.228.202:31316/test mysql.username=devrdbusr21 mysql.password=devrdbusr21 -mysql.driverclass=com.mysql.jdbc.Driver \ No newline at end of file +mysql.driver_class=com.mysql.jdbc.Driver + +filter.filter_package_name=com.ai.cloud.skywalking.analysis.filter.impl + +chainnodesummary.interval=5 \ No newline at end of file diff --git a/skywalking-analysis/src/test/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapperTest.java b/skywalking-analysis/src/test/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapperTest.java index f88e09197..7d6475150 100644 --- a/skywalking-analysis/src/test/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapperTest.java +++ b/skywalking-analysis/src/test/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapperTest.java @@ -2,7 +2,7 @@ package com.ai.cloud.skywalking.analysis.mapper; import com.ai.cloud.skywalking.analysis.categorize2chain.Categorize2ChainMapper; -import com.ai.cloud.skywalking.analysis.categorize2chain.Categorize2ChainReduce; +import com.ai.cloud.skywalking.analysis.categorize2chain.Categorize2ChainReducer; import com.ai.cloud.skywalking.analysis.categorize2chain.model.ChainInfo; import com.ai.cloud.skywalking.analysis.config.Config; import com.ai.cloud.skywalking.analysis.config.ConfigInitializer; @@ -43,12 +43,12 @@ public class CallChainMapperTest { List chainInfos = new ArrayList(); chainInfos.add(chainInfo); - Categorize2ChainReduce.reduceAction(chainInfo.getUserId() + ":" + chainInfo.getEntranceNodeToken(), chainInfos.iterator()); + Categorize2ChainReducer.reduceAction(chainInfo.getUserId() + ":" + chainInfo.getEntranceNodeToken(), chainInfos.iterator()); } public static List selectByTraceId(String traceId) throws IOException { List entries = new ArrayList(); - Table table = connection.getTable(TableName.valueOf(Config.HBase.CALL_CHAIN_TABLE_NAME)); + Table table = connection.getTable(TableName.valueOf(Config.HBase.TABLE_CALL_CHAIN)); Get g = new Get(Bytes.toBytes(traceId)); Result r = table.get(g); for (Cell cell : r.rawCells()) {