Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
apache
SkyWalking
提交
24fd3ff1
S
SkyWalking
项目概览
apache
/
SkyWalking
上一次同步 1 年多
通知
302
Star
21345
Fork
6091
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
S
SkyWalking
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
24fd3ff1
编写于
1月 14, 2016
作者:
Z
zhangxin10
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
1. 完成ViewPoint的链
2. 完成处理逻辑的链
上级
1d1ba9d1
变更
23
隐藏空白更改
内联
并排
Showing
23 changed file
with
853 addition
and
0 deletion
+853
-0
skywalking-analysis/pom.xml
skywalking-analysis/pom.xml
+66
-0
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/AnalysisServerDriver.java
...om/ai/cloud/skywalking/analysis/AnalysisServerDriver.java
+58
-0
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/Config.java
.../java/com/ai/cloud/skywalking/analysis/config/Config.java
+7
-0
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/ConfigInitializer.java
...i/cloud/skywalking/analysis/config/ConfigInitializer.java
+85
-0
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/ViewPointFilterFactory.java
...ud/skywalking/analysis/config/ViewPointFilterFactory.java
+79
-0
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/ChainNodeFilter.java
.../ai/cloud/skywalking/analysis/filter/ChainNodeFilter.java
+9
-0
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/NodeChain.java
...va/com/ai/cloud/skywalking/analysis/filter/NodeChain.java
+37
-0
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/CopyAttrNodeFilter.java
...d/skywalking/analysis/filter/impl/CopyAttrNodeFilter.java
+26
-0
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/CostNodeFilter.java
...cloud/skywalking/analysis/filter/impl/CostNodeFilter.java
+24
-0
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/ViewPointNodeFilter.java
.../skywalking/analysis/filter/impl/ViewPointNodeFilter.java
+20
-0
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapper.java
.../ai/cloud/skywalking/analysis/mapper/CallChainMapper.java
+79
-0
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainInfo.java
...ava/com/ai/cloud/skywalking/analysis/model/ChainInfo.java
+105
-0
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainNode.java
...ava/com/ai/cloud/skywalking/analysis/model/ChainNode.java
+97
-0
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/CostMap.java
.../java/com/ai/cloud/skywalking/analysis/model/CostMap.java
+20
-0
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/ChainInfoReduce.java
.../ai/cloud/skywalking/analysis/reduce/ChainInfoReduce.java
+15
-0
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/util/TokenGenerator.java
...com/ai/cloud/skywalking/analysis/util/TokenGenerator.java
+36
-0
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/viewpoint/ViewPointFilter.java
.../cloud/skywalking/analysis/viewpoint/ViewPointFilter.java
+18
-0
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/viewpoint/impl/AppendBusinessKeyFilter.java
...king/analysis/viewpoint/impl/AppendBusinessKeyFilter.java
+15
-0
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/viewpoint/impl/ReplaceIPPortFilter.java
...ywalking/analysis/viewpoint/impl/ReplaceIPPortFilter.java
+22
-0
skywalking-analysis/src/main/resources/config.properties
skywalking-analysis/src/main/resources/config.properties
+0
-0
skywalking-analysis/src/main/resources/log4j.xml
skywalking-analysis/src/main/resources/log4j.xml
+19
-0
skywalking-analysis/src/main/resources/log4j2.xml
skywalking-analysis/src/main/resources/log4j2.xml
+13
-0
skywalking-analysis/src/main/resources/viewpointfilter.conf
skywalking-analysis/src/main/resources/viewpointfilter.conf
+3
-0
未找到文件。
skywalking-analysis/pom.xml
0 → 100644
浏览文件 @
24fd3ff1
<project
xmlns=
"http://maven.apache.org/POM/4.0.0"
xmlns:xsi=
"http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation=
"http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
>
<modelVersion>
4.0.0
</modelVersion>
<groupId>
com.ai.cloud
</groupId>
<artifactId>
skywalking-analysis
</artifactId>
<version>
1.0-SNAPSHOT
</version>
<packaging>
jar
</packaging>
<name>
skywalking-analysis
</name>
<url>
http://maven.apache.org
</url>
<properties>
<project.build.sourceEncoding>
UTF-8
</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>
junit
</groupId>
<artifactId>
junit
</artifactId>
<version>
3.8.1
</version>
<scope>
test
</scope>
</dependency>
<dependency>
<groupId>
org.apache.hbase
</groupId>
<artifactId>
hbase-client
</artifactId>
<version>
1.1.2
</version>
</dependency>
<dependency>
<groupId>
org.apache.hbase
</groupId>
<artifactId>
hbase-server
</artifactId>
<version>
1.1.2
</version>
</dependency>
<dependency>
<groupId>
org.apache.logging.log4j
</groupId>
<artifactId>
log4j-core
</artifactId>
<version>
2.4.1
</version>
</dependency>
<dependency>
<groupId>
com.ai.cloud
</groupId>
<artifactId>
skywalking-protocol
</artifactId>
<version>
1.0-SNAPSHOT
</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<artifactId>
maven-compiler-plugin
</artifactId>
<configuration>
<source>
1.6
</source>
<target>
1.6
</target>
<encoding>
${project.build.sourceEncoding}
</encoding>
</configuration>
</plugin>
<plugin>
<groupId>
org.apache.maven.plugins
</groupId>
<artifactId>
maven-resources-plugin
</artifactId>
<version>
2.4.3
</version>
<configuration>
<encoding>
${project.build.sourceEncoding}
</encoding>
</configuration>
</plugin>
</plugins>
</build>
</project>
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/AnalysisServerDriver.java
0 → 100644
浏览文件 @
24fd3ff1
package
com.ai.cloud.skywalking.analysis
;
import
com.ai.cloud.skywalking.analysis.config.Config
;
import
com.ai.cloud.skywalking.analysis.mapper.CallChainMapper
;
import
com.ai.cloud.skywalking.analysis.model.ChainInfo
;
import
org.apache.hadoop.conf.Configuration
;
import
org.apache.hadoop.conf.Configured
;
import
org.apache.hadoop.hbase.client.Scan
;
import
org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil
;
import
org.apache.hadoop.mapreduce.Job
;
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
java.io.IOException
;
import
java.text.ParseException
;
import
java.text.SimpleDateFormat
;
import
java.util.Date
;
public
class
AnalysisServerDriver
extends
Configured
implements
Tool
{
private
static
Logger
logger
=
LoggerFactory
.
getLogger
(
AnalysisServerDriver
.
class
.
getName
());
public
static
void
main
(
String
[]
args
)
throws
Exception
{
logger
.
info
(
"Begin to analysis call chain."
);
int
res
=
ToolRunner
.
run
(
new
AnalysisServerDriver
(),
args
);
System
.
exit
(
res
);
}
@Override
public
int
run
(
String
[]
args
)
throws
Exception
{
Configuration
conf
=
new
Configuration
();
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
.
TABLE_NAME
,
scan
,
CallChainMapper
.
class
,
String
.
class
,
ChainInfo
.
class
,
job
);
//TableMapReduceUtil.initTableReducerJob("sw-call-chain-model", CallChainReducer.class, job);
return
job
.
waitForCompletion
(
true
)
?
0
:
1
;
}
private
Scan
buildHBaseScan
(
String
[]
args
)
throws
ParseException
,
IOException
{
SimpleDateFormat
simpleDateFormat
=
new
SimpleDateFormat
(
"yyyy-MM-dd/HH:mm:ss"
);
Date
startDate
=
simpleDateFormat
.
parse
(
args
[
0
]);
Date
endDate
=
simpleDateFormat
.
parse
(
args
[
1
]);
Scan
scan
=
new
Scan
();
scan
.
setTimeRange
(
startDate
.
getTime
(),
endDate
.
getTime
());
return
scan
;
}
}
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/Config.java
0 → 100644
浏览文件 @
24fd3ff1
package
com.ai.cloud.skywalking.analysis.config
;
public
class
Config
{
public
static
class
HBase
{
public
static
String
TABLE_NAME
;
}
}
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/ConfigInitializer.java
0 → 100644
浏览文件 @
24fd3ff1
package
com.ai.cloud.skywalking.analysis.config
;
import
java.io.IOException
;
import
java.io.InputStream
;
import
java.lang.reflect.Field
;
import
java.lang.reflect.Modifier
;
import
java.util.LinkedList
;
import
java.util.Properties
;
import
java.util.logging.Level
;
import
java.util.logging.Logger
;
public
class
ConfigInitializer
{
private
static
Logger
logger
=
Logger
.
getLogger
(
ConfigInitializer
.
class
.
getName
());
public
static
void
initialize
()
{
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
{
try
{
Properties
properties
=
new
Properties
();
properties
.
load
(
inputStream
);
initNextLevel
(
properties
,
Config
.
class
,
new
ConfigDesc
());
}
catch
(
IllegalAccessException
e
)
{
logger
.
log
(
Level
.
ALL
,
"Parsing certification file failed, sky-walking api auto shutdown."
);
}
catch
(
IOException
e
)
{
logger
.
log
(
Level
.
ALL
,
"Failed to read the certification file, sky-walking api auto shutdown."
);
}
}
}
private
static
void
initNextLevel
(
Properties
properties
,
Class
<?>
recentConfigType
,
ConfigDesc
parentDesc
)
throws
NumberFormatException
,
IllegalArgumentException
,
IllegalAccessException
{
for
(
Field
field
:
recentConfigType
.
getFields
())
{
if
(
Modifier
.
isPublic
(
field
.
getModifiers
())
&&
Modifier
.
isStatic
(
field
.
getModifiers
()))
{
String
configKey
=
(
parentDesc
+
"."
+
field
.
getName
()).
toLowerCase
();
String
value
=
properties
.
getProperty
(
configKey
);
if
(
value
!=
null
)
{
if
(
field
.
getType
().
equals
(
int
.
class
))
field
.
set
(
null
,
Integer
.
valueOf
(
value
));
if
(
field
.
getType
().
equals
(
String
.
class
))
field
.
set
(
null
,
value
);
if
(
field
.
getType
().
equals
(
long
.
class
))
field
.
set
(
null
,
Long
.
valueOf
(
value
));
if
(
field
.
getType
().
equals
(
boolean
.
class
))
field
.
set
(
null
,
Boolean
.
valueOf
(
value
));
}
}
}
for
(
Class
<?>
innerConfiguration
:
recentConfigType
.
getClasses
())
{
parentDesc
.
append
(
innerConfiguration
.
getSimpleName
());
initNextLevel
(
properties
,
innerConfiguration
,
parentDesc
);
parentDesc
.
removeLastDesc
();
}
}
}
class
ConfigDesc
{
private
LinkedList
<
String
>
descs
=
new
LinkedList
<
String
>();
void
append
(
String
currentDesc
)
{
descs
.
addLast
(
currentDesc
);
}
void
removeLastDesc
()
{
descs
.
removeLast
();
}
@Override
public
String
toString
()
{
if
(
descs
.
size
()
==
0
)
{
return
""
;
}
StringBuilder
ret
=
new
StringBuilder
(
descs
.
getFirst
());
boolean
first
=
true
;
for
(
String
desc
:
descs
)
{
if
(
first
)
{
first
=
false
;
continue
;
}
ret
.
append
(
"."
).
append
(
desc
);
}
return
ret
.
toString
();
}
}
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/ViewPointFilterFactory.java
0 → 100644
浏览文件 @
24fd3ff1
package
com.ai.cloud.skywalking.analysis.config
;
import
com.ai.cloud.skywalking.analysis.viewpoint.ViewPointFilter
;
import
org.slf4j.Logger
;
import
org.slf4j.LoggerFactory
;
import
java.io.IOException
;
import
java.util.HashMap
;
import
java.util.Map
;
import
java.util.Properties
;
import
java.util.Set
;
public
class
ViewPointFilterFactory
{
private
static
Logger
logger
=
LoggerFactory
.
getLogger
(
ViewPointFilterFactory
.
class
.
getName
());
private
static
Map
<
String
,
ViewPointFilter
>
filterType
;
private
ViewPointFilterFactory
()
{
//Non
}
private
static
void
initFilterChain
()
{
Properties
properties
=
new
Properties
();
try
{
properties
.
load
(
ViewPointFilterFactory
.
class
.
getResourceAsStream
(
"/viewpointfilter.conf"
));
}
catch
(
IOException
e
)
{
logger
.
error
(
"Failed to found the conf file[viewpointfilter.conf]"
,
e
);
System
.
exit
(-
1
);
}
Set
<
Map
.
Entry
<
Object
,
Object
>>
entries
=
properties
.
entrySet
();
for
(
Map
.
Entry
<
Object
,
Object
>
entry
:
entries
)
{
String
types
=
(
String
)
entry
.
getKey
();
String
filters
=
(
String
)
entry
.
getValue
();
String
[]
filterClasses
=
filters
.
split
(
">"
);
ViewPointFilter
filter
=
null
;
for
(
int
i
=
filterClasses
.
length
-
1
;
i
>=
0
;
i
--)
{
try
{
Class
filterClass
=
Class
.
forName
(
filterClasses
[
i
]);
ViewPointFilter
newFilter
=
(
ViewPointFilter
)
filterClass
.
newInstance
();
newFilter
.
setViewPointFilter
(
filter
);
filter
=
newFilter
;
}
catch
(
ClassNotFoundException
e
)
{
logger
.
error
(
"Failed to found class["
+
filterClasses
[
i
]
+
"]."
,
e
);
System
.
exit
(-
1
);
}
catch
(
InstantiationException
e
)
{
logger
.
error
(
"Failed to instance class["
+
filterClasses
[
i
]
+
"]."
,
e
);
System
.
exit
(-
1
);
}
catch
(
IllegalAccessException
e
)
{
logger
.
error
(
"Failed to access class["
+
filterClasses
[
i
]
+
"]."
,
e
);
System
.
exit
(-
1
);
}
}
filterType
=
new
HashMap
<
String
,
ViewPointFilter
>();
String
[]
type
=
types
.
split
(
","
);
for
(
int
i
=
0
;
i
<
type
.
length
;
i
++)
{
filterType
.
put
(
type
[
i
],
filter
);
}
}
}
public
static
ViewPointFilter
getFilter
(
String
type
)
{
if
(
filterType
==
null
)
{
initFilterChain
();
}
ViewPointFilter
filter
=
filterType
.
get
(
type
);
if
(
filter
==
null
)
{
throw
new
RuntimeException
(
"Failed to found the filter["
+
type
+
"]"
);
}
return
filter
;
}
}
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/ChainNodeFilter.java
0 → 100644
浏览文件 @
24fd3ff1
package
com.ai.cloud.skywalking.analysis.filter
;
import
com.ai.cloud.skywalking.analysis.model.ChainNode
;
import
com.ai.cloud.skywalking.analysis.model.CostMap
;
import
com.ai.cloud.skywalking.protocol.Span
;
public
interface
ChainNodeFilter
{
void
doFilter
(
Span
span
,
ChainNode
node
,
CostMap
costMap
,
NodeChain
chain
);
}
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/NodeChain.java
0 → 100644
浏览文件 @
24fd3ff1
package
com.ai.cloud.skywalking.analysis.filter
;
import
com.ai.cloud.skywalking.analysis.filter.impl.CopyAttrNodeFilter
;
import
com.ai.cloud.skywalking.analysis.filter.impl.CostNodeFilter
;
import
com.ai.cloud.skywalking.analysis.filter.impl.ViewPointNodeFilter
;
import
com.ai.cloud.skywalking.analysis.model.ChainNode
;
import
com.ai.cloud.skywalking.analysis.model.CostMap
;
import
com.ai.cloud.skywalking.protocol.Span
;
import
java.util.ArrayList
;
import
java.util.List
;
public
class
NodeChain
{
private
int
stage
=
0
;
private
static
List
<
ChainNodeFilter
>
filters
=
null
;
public
void
doChain
(
Span
span
,
ChainNode
chainNode
,
CostMap
costMap
)
{
if
(
filters
==
null
)
{
filters
=
new
ArrayList
<
ChainNodeFilter
>();
init
();
}
int
subscript
=
stage
;
stage
=
stage
+
1
;
if
(
subscript
<
filters
.
size
())
{
filters
.
get
(
subscript
).
doFilter
(
span
,
chainNode
,
costMap
,
this
);
}
}
private
static
void
init
()
{
filters
.
add
(
new
CostNodeFilter
());
filters
.
add
(
new
CopyAttrNodeFilter
());
filters
.
add
(
new
ViewPointNodeFilter
());
}
}
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/CopyAttrNodeFilter.java
0 → 100644
浏览文件 @
24fd3ff1
package
com.ai.cloud.skywalking.analysis.filter.impl
;
import
com.ai.cloud.skywalking.analysis.filter.ChainNodeFilter
;
import
com.ai.cloud.skywalking.analysis.filter.NodeChain
;
import
com.ai.cloud.skywalking.analysis.model.ChainNode
;
import
com.ai.cloud.skywalking.analysis.model.CostMap
;
import
com.ai.cloud.skywalking.protocol.Span
;
public
class
CopyAttrNodeFilter
implements
ChainNodeFilter
{
@Override
public
void
doFilter
(
Span
span
,
ChainNode
node
,
CostMap
costMap
,
NodeChain
chain
)
{
node
.
setCost
(
span
.
getCost
());
node
.
setLevelId
(
span
.
getLevelId
());
node
.
setParentLevelId
(
span
.
getParentLevel
());
node
.
setCallType
(
span
.
getCallType
());
if
(
span
.
getExceptionStack
()
==
null
||
span
.
getExceptionStack
().
length
()
==
0
)
{
node
.
setStatus
(
ChainNode
.
NodeStatus
.
NORMAL
);
}
else
{
node
.
setStatus
(
ChainNode
.
NodeStatus
.
ABNORMAL
);
}
chain
.
doChain
(
span
,
node
,
costMap
);
}
}
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/CostNodeFilter.java
0 → 100644
浏览文件 @
24fd3ff1
package
com.ai.cloud.skywalking.analysis.filter.impl
;
import
com.ai.cloud.skywalking.analysis.filter.ChainNodeFilter
;
import
com.ai.cloud.skywalking.analysis.filter.NodeChain
;
import
com.ai.cloud.skywalking.analysis.model.ChainNode
;
import
com.ai.cloud.skywalking.analysis.model.CostMap
;
import
com.ai.cloud.skywalking.protocol.Span
;
public
class
CostNodeFilter
implements
ChainNodeFilter
{
@Override
public
void
doFilter
(
Span
span
,
ChainNode
node
,
CostMap
costMap
,
NodeChain
chain
)
{
if
(
span
.
isReceiver
())
{
costMap
.
put
(
span
.
getParentLevel
()
+
"."
+
span
.
getLevelId
()
+
"-S"
,
node
.
getCost
());
}
else
{
if
(
costMap
.
exists
(
span
.
getParentLevel
()))
{
costMap
.
put
(
span
.
getParentLevel
(),
costMap
.
get
(
span
.
getParentLevel
())
+
span
.
getCost
());
}
else
{
costMap
.
put
(
span
.
getParentLevel
(),
span
.
getCost
());
}
chain
.
doChain
(
span
,
node
,
costMap
);
}
}
}
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/filter/impl/ViewPointNodeFilter.java
0 → 100644
浏览文件 @
24fd3ff1
package
com.ai.cloud.skywalking.analysis.filter.impl
;
import
com.ai.cloud.skywalking.analysis.config.ViewPointFilterFactory
;
import
com.ai.cloud.skywalking.analysis.filter.ChainNodeFilter
;
import
com.ai.cloud.skywalking.analysis.filter.NodeChain
;
import
com.ai.cloud.skywalking.analysis.model.ChainNode
;
import
com.ai.cloud.skywalking.analysis.model.CostMap
;
import
com.ai.cloud.skywalking.analysis.viewpoint.ViewPointFilter
;
import
com.ai.cloud.skywalking.protocol.Span
;
public
class
ViewPointNodeFilter
implements
ChainNodeFilter
{
@Override
public
void
doFilter
(
Span
span
,
ChainNode
node
,
CostMap
costMap
,
NodeChain
chain
)
{
ViewPointFilter
viewPointFilter
=
ViewPointFilterFactory
.
getFilter
(
span
.
getSpanType
());
String
viewPoint
=
span
.
getViewPointId
();
viewPointFilter
.
doFilter
(
span
,
viewPoint
);
node
.
setViewPoint
(
viewPoint
);
chain
.
doChain
(
span
,
node
,
costMap
);
}
}
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapper.java
0 → 100644
浏览文件 @
24fd3ff1
package
com.ai.cloud.skywalking.analysis.mapper
;
import
com.ai.cloud.skywalking.analysis.model.ChainInfo
;
import
com.ai.cloud.skywalking.analysis.model.ChainNode
;
import
com.ai.cloud.skywalking.protocol.CallType
;
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.slf4j.Logger
;
import
org.slf4j.LoggerFactory
;
import
java.io.IOException
;
import
java.util.HashMap
;
import
java.util.Map
;
public
class
CallChainMapper
extends
TableMapper
<
Text
,
ChainInfo
>
{
private
Logger
logger
=
LoggerFactory
.
getLogger
(
CallChainMapper
.
class
.
getName
());
@Override
protected
void
map
(
ImmutableBytesWritable
key
,
Result
value
,
Context
context
)
throws
IOException
,
InterruptedException
{
Map
<
String
,
Long
>
costMap
=
new
HashMap
<
String
,
Long
>();
ChainInfo
chainInfo
=
new
ChainInfo
();
String
traceId
=
key
.
toString
();
logger
.
info
(
"Begin to deal Trace["
+
traceId
+
"]..."
);
for
(
Cell
cell
:
value
.
rawCells
())
{
ChainNode
node
=
new
ChainNode
();
Span
span
=
new
Span
(
Bytes
.
toString
(
cell
.
getValueArray
(),
cell
.
getValueOffset
(),
cell
.
getValueLength
()));
node
.
setLevelId
(
span
.
getLevelId
());
node
.
setParentLevelId
(
span
.
getParentLevel
());
node
.
setViewPoint
(
convertViewPoint
(
span
));
node
.
setCost
(
span
.
getCost
());
node
.
setCallType
(
span
.
getCallType
());
if
(
span
.
isReceiver
())
{
costMap
.
put
(
span
.
getParentLevel
()
+
"."
+
span
.
getLevelId
()
+
"-S"
,
span
.
getCost
());
continue
;
}
if
(
costMap
.
containsKey
(
span
.
getParentLevel
()))
{
costMap
.
put
(
span
.
getParentLevel
(),
costMap
.
get
(
span
.
getParentLevel
()).
longValue
()
+
span
.
getCost
());
}
else
{
costMap
.
put
(
span
.
getParentLevel
(),
0L
);
}
chainInfo
.
getNodes
().
add
(
node
);
}
for
(
ChainNode
node
:
chainInfo
.
getNodes
())
{
CallType
callType
=
CallType
.
valueOf
(
node
.
getCallType
());
switch
(
callType
)
{
case
ASYNC:
break
;
case
SYNC:
break
;
case
LOCAL:
break
;
}
}
//入Mysql库
//入HBase库
//找到首个Node的MD5
//拼接MR任务的Key
context
.
write
(
new
Text
(
""
),
chainInfo
);
}
private
String
convertViewPoint
(
Span
span
)
{
return
null
;
}
}
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainInfo.java
0 → 100644
浏览文件 @
24fd3ff1
package
com.ai.cloud.skywalking.analysis.model
;
import
org.apache.hadoop.io.Writable
;
import
java.io.DataInput
;
import
java.io.DataOutput
;
import
java.io.IOException
;
import
java.util.ArrayList
;
import
java.util.List
;
public
class
ChainInfo
implements
Writable
{
private
String
chainToken
;
private
ChainStatus
chainStatus
;
private
List
<
ChainNode
>
nodes
;
public
ChainInfo
()
{
this
.
nodes
=
new
ArrayList
<
ChainNode
>();
}
@Override
public
void
write
(
DataOutput
out
)
throws
IOException
{
out
.
write
(
chainToken
.
getBytes
());
out
.
writeChar
(
chainStatus
.
getValue
());
out
.
writeInt
(
nodes
.
size
());
for
(
ChainNode
chainNode
:
nodes
)
{
out
.
write
(
chainNode
.
getNodeToken
().
getBytes
());
out
.
write
(
chainNode
.
getViewPoint
().
getBytes
());
out
.
writeChar
(
chainNode
.
getStatus
().
getValue
());
out
.
writeLong
(
chainNode
.
getCost
());
out
.
write
(
chainNode
.
getParentLevelId
().
getBytes
());
out
.
writeInt
(
chainNode
.
getLevelId
());
}
}
@Override
public
void
readFields
(
DataInput
in
)
throws
IOException
{
chainToken
=
in
.
readLine
();
chainStatus
=
ChainStatus
.
convert
(
in
.
readChar
());
int
nodeSize
=
in
.
readInt
();
this
.
nodes
=
new
ArrayList
<
ChainNode
>();
for
(
int
i
=
0
;
i
<
nodeSize
;
i
++)
{
ChainNode
chainNode
=
new
ChainNode
();
chainNode
.
setNodeToken
(
in
.
readLine
());
chainNode
.
setViewPoint
(
in
.
readLine
());
chainNode
.
setStatus
(
ChainNode
.
NodeStatus
.
convert
(
in
.
readChar
()));
chainNode
.
setCost
(
in
.
readLong
());
chainNode
.
setParentLevelId
(
in
.
readLine
());
chainNode
.
setLevelId
(
in
.
readInt
());
nodes
.
add
(
chainNode
);
}
}
public
String
getChainToken
()
{
return
chainToken
;
}
public
void
setChainToken
(
String
chainToken
)
{
this
.
chainToken
=
chainToken
;
}
public
ChainStatus
getChainStatus
()
{
return
chainStatus
;
}
public
void
setChainStatus
(
ChainStatus
chainStatus
)
{
this
.
chainStatus
=
chainStatus
;
}
public
List
<
ChainNode
>
getNodes
()
{
return
nodes
;
}
public
void
setNodes
(
List
<
ChainNode
>
nodes
)
{
this
.
nodes
=
nodes
;
}
public
enum
ChainStatus
{
NORMAL
(
'N'
),
ABNORMAL
(
'A'
);
private
char
value
;
ChainStatus
(
char
value
)
{
this
.
value
=
value
;
}
public
char
getValue
()
{
return
value
;
}
public
static
ChainStatus
convert
(
char
value
)
{
switch
(
value
)
{
case
'N'
:
return
NORMAL
;
case
'A'
:
return
ABNORMAL
;
default
:
throw
new
IllegalStateException
(
"Failed to convert["
+
value
+
"]"
);
}
}
}
}
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/ChainNode.java
0 → 100644
浏览文件 @
24fd3ff1
package
com.ai.cloud.skywalking.analysis.model
;
public
class
ChainNode
{
private
String
nodeToken
;
private
String
viewPoint
;
//TODO
private
String
businessKey
;
private
long
cost
;
private
NodeStatus
status
;
private
String
parentLevelId
;
private
int
levelId
;
private
String
callType
;
public
String
getNodeToken
()
{
return
nodeToken
;
}
public
void
setNodeToken
(
String
nodeToken
)
{
this
.
nodeToken
=
nodeToken
;
}
public
String
getViewPoint
()
{
return
viewPoint
;
}
public
void
setViewPoint
(
String
viewPoint
)
{
this
.
viewPoint
=
viewPoint
;
}
public
long
getCost
()
{
return
cost
;
}
public
void
setCost
(
long
cost
)
{
this
.
cost
=
cost
;
}
public
NodeStatus
getStatus
()
{
return
status
;
}
public
void
setStatus
(
NodeStatus
status
)
{
this
.
status
=
status
;
}
public
String
getParentLevelId
()
{
return
parentLevelId
;
}
public
void
setParentLevelId
(
String
parentLevelId
)
{
this
.
parentLevelId
=
parentLevelId
;
}
public
int
getLevelId
()
{
return
levelId
;
}
public
void
setLevelId
(
int
levelId
)
{
this
.
levelId
=
levelId
;
}
public
void
setCallType
(
String
callType
)
{
this
.
callType
=
callType
;
}
public
String
getCallType
()
{
return
callType
;
}
public
enum
NodeStatus
{
NORMAL
(
'N'
),
ABNORMAL
(
'A'
);
private
char
value
;
NodeStatus
(
char
value
)
{
this
.
value
=
value
;
}
public
char
getValue
()
{
return
value
;
}
public
static
NodeStatus
convert
(
char
value
)
{
switch
(
value
)
{
case
'N'
:
return
NORMAL
;
case
'A'
:
return
ABNORMAL
;
default
:
throw
new
IllegalStateException
(
"Failed to convert["
+
value
+
"]"
);
}
}
}
}
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/model/CostMap.java
0 → 100644
浏览文件 @
24fd3ff1
package
com.ai.cloud.skywalking.analysis.model
;
import
java.util.HashMap
;
import
java.util.Map
;
public
class
CostMap
{
private
Map
<
String
,
Long
>
costs
=
new
HashMap
<
String
,
Long
>();
public
void
put
(
String
parentLevel
,
Long
cost
)
{
costs
.
put
(
parentLevel
,
cost
);
}
public
boolean
exists
(
String
parentLevel
)
{
return
costs
.
containsKey
(
parentLevel
);
}
public
long
get
(
String
parentLevel
)
{
return
costs
.
get
(
parentLevel
);
}
}
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/reduce/ChainInfoReduce.java
0 → 100644
浏览文件 @
24fd3ff1
package
com.ai.cloud.skywalking.analysis.reduce
;
import
com.ai.cloud.skywalking.analysis.model.ChainInfo
;
import
org.apache.hadoop.hbase.client.Put
;
import
org.apache.hadoop.hbase.mapreduce.TableReducer
;
import
org.apache.hadoop.io.Text
;
import
java.io.IOException
;
public
class
ChainInfoReduce
extends
TableReducer
<
Text
,
ChainInfo
,
Put
>
{
@Override
protected
void
reduce
(
Text
key
,
Iterable
<
ChainInfo
>
values
,
Context
context
)
throws
IOException
,
InterruptedException
{
super
.
reduce
(
key
,
values
,
context
);
}
}
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/util/TokenGenerator.java
0 → 100644
浏览文件 @
24fd3ff1
package
com.ai.cloud.skywalking.analysis.util
;
import
org.slf4j.Logger
;
import
org.slf4j.LoggerFactory
;
import
java.security.MessageDigest
;
import
java.security.NoSuchAlgorithmException
;
public
class
TokenGenerator
{
private
static
Logger
logger
=
LoggerFactory
.
getLogger
(
TokenGenerator
.
class
.
getName
());
private
TokenGenerator
()
{
//Non
}
public
static
String
generate
(
String
originData
)
{
StringBuilder
result
=
new
StringBuilder
();
if
(
originData
!=
null
)
{
try
{
MessageDigest
md
=
MessageDigest
.
getInstance
(
"MD5"
);
byte
bytes
[]
=
md
.
digest
(
originData
.
getBytes
());
for
(
int
i
=
0
;
i
<
bytes
.
length
;
i
++)
{
String
str
=
Integer
.
toHexString
(
bytes
[
i
]
&
0xFF
);
if
(
str
.
length
()
==
1
)
{
str
+=
"F"
;
}
result
.
append
(
str
);
}
}
catch
(
NoSuchAlgorithmException
e
)
{
logger
.
error
(
"Cannot found algorithm."
,
e
);
System
.
exit
(-
1
);
}
}
return
result
.
toString
().
toUpperCase
();
}
}
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/viewpoint/ViewPointFilter.java
0 → 100644
浏览文件 @
24fd3ff1
package
com.ai.cloud.skywalking.analysis.viewpoint
;
import
com.ai.cloud.skywalking.protocol.Span
;
public
abstract
class
ViewPointFilter
{
private
ViewPointFilter
viewPointFilter
;
public
abstract
void
doFilter
(
Span
viewPoint
,
String
span
);
public
ViewPointFilter
getViewPointFilter
()
{
return
viewPointFilter
;
}
public
void
setViewPointFilter
(
ViewPointFilter
viewPointFilter
)
{
this
.
viewPointFilter
=
viewPointFilter
;
}
}
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/viewpoint/impl/AppendBusinessKeyFilter.java
0 → 100644
浏览文件 @
24fd3ff1
package
com.ai.cloud.skywalking.analysis.viewpoint.impl
;
import
com.ai.cloud.skywalking.analysis.viewpoint.ViewPointFilter
;
import
com.ai.cloud.skywalking.protocol.Span
;
public
class
AppendBusinessKeyFilter
extends
ViewPointFilter
{
@Override
public
void
doFilter
(
Span
span
,
String
viewPoint
)
{
viewPoint
+=
span
.
getBusinessKey
();
if
(
getViewPointFilter
()
!=
null
)
{
getViewPointFilter
().
doFilter
(
span
,
viewPoint
);
}
}
}
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/viewpoint/impl/ReplaceIPPortFilter.java
0 → 100644
浏览文件 @
24fd3ff1
package
com.ai.cloud.skywalking.analysis.viewpoint.impl
;
import
com.ai.cloud.skywalking.analysis.viewpoint.ViewPointFilter
;
import
com.ai.cloud.skywalking.protocol.Span
;
public
class
ReplaceIPPortFilter
extends
ViewPointFilter
{
//ip:port regex
private
static
String
IP_PORT_REGEX
=
"^([0-9]|[1-9][0-9]|1([0-9][0-9])|2([0-4][0-9]|5[0-5]))\\."
+
"([0-9]|[1-9][0-9]|1([0-9][0-9])|2([0-4][0-9]|5[0-5]))\\.([0-9]|[1-9][0-9]|1([0-9][0-9])"
+
"|2([0-4][0-9]|5[0-5]))\\.([0-9]|[1-9][0-9]|1([0-9][0-9])|2([0-4][0-9]|5[0-5])):"
+
"([0-9]{1,4}|[1-5][0-9]{4}|6[0-4][0-9]{3}|65[0-4][0-9]{2}|655[0-2][0-9]|6553[0-5])$"
;
@Override
public
void
doFilter
(
Span
span
,
String
viewPoint
)
{
viewPoint
=
viewPoint
.
replaceAll
(
IP_PORT_REGEX
,
span
.
getApplicationId
());
if
(
getViewPointFilter
()
!=
null
)
{
getViewPointFilter
().
doFilter
(
span
,
viewPoint
);
}
}
}
skywalking-analysis/src/main/resources/config.properties
0 → 100644
浏览文件 @
24fd3ff1
skywalking-analysis/src/main/resources/log4j.xml
0 → 100644
浏览文件 @
24fd3ff1
<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE log4j:configuration SYSTEM "log4j.dtd">
<log4j:configuration>
<appender
name=
"CONSOLE"
class=
"org.apache.log4j.ConsoleAppender"
>
<layout
class=
"org.apache.log4j.PatternLayout"
>
<param
name=
"ConversionPattern"
value=
"%d - %c -%-4r [%t] %-5p %x - %m%n"
/>
</layout>
</appender>
<root>
<priority
value=
"debug"
/>
<appender-ref
ref=
"CONSOLE"
/>
</root>
</log4j:configuration>
skywalking-analysis/src/main/resources/log4j2.xml
0 → 100644
浏览文件 @
24fd3ff1
<?xml version="1.0" encoding="UTF-8"?>
<Configuration
status=
"debug"
>
<Appenders>
<Console
name=
"Console"
target=
"SYSTEM_OUT"
>
<PatternLayout
pattern=
"%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"
/>
</Console>
</Appenders>
<Loggers>
<Root
level=
"debug"
>
<AppenderRef
ref=
"Console"
/>
</Root>
</Loggers>
</Configuration>
\ No newline at end of file
skywalking-analysis/src/main/resources/viewpointfilter.conf
0 → 100644
浏览文件 @
24fd3ff1
#
D
,
M
=
com
.
ai
.
cloud
.
skywalking
.
analysis
.
viewpoint
.
impl
.
ReplaceIPPortFilter
J
=
com
.
ai
.
cloud
.
skywalking
.
analysis
.
viewpoint
.
impl
.
AppendBusinessKeyFilter
\ No newline at end of file
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录