gfwleak/galaxy-tsg-olap-flume-dynamic-complement
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Flume Dynamic_Complement Initial Commit 202006121927

Browse Source
This commit is contained in:
caohui
2020-06-12 19:28:37 +08:00
commit f8dfa376c3
13 changed files with 1753 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
.idea/compiler.xml generated Normal file
View File

@@ -0,0 +1,13 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="CompilerConfiguration">
<annotationProcessing>
<profile name="Maven default annotation processors profile" enabled="true">
<sourceOutputDir name="target/generated-sources/annotations" />
<sourceTestOutputDir name="target/generated-test-sources/test-annotations" />
<outputRelativeToContentRoot value="true" />
<module name="FlumeDynamicInterceptor" />
</profile>
</annotationProcessing>
</component>
</project>

6
.idea/encodings.xml generated Normal file
View File

@@ -0,0 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="Encoding">
<file url="file://$PROJECT_DIR$/FlumeDynamicInterceptor" charset="UTF-8" />
</component>
</project>

14
.idea/misc.xml generated Normal file
View File

@@ -0,0 +1,14 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="ExternalStorageConfigurationManager" enabled="true" />
<component name="MavenProjectsManager">
<option name="originalFiles">
<list>
<option value="$PROJECT_DIR$/pom.xml" />
</list>
</option>
</component>
<component name="ProjectRootManager" version="2" languageLevel="JDK_1_8" project-jdk-name="1.8" project-jdk-type="JavaSDK">
<output url="file://$PROJECT_DIR$/out" />
</component>
</project>

462
.idea/workspace.xml generated Normal file
View File

@@ -0,0 +1,462 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="ChangeListManager">
<list default="true" id="7803fbe0-0492-4e54-9d86-e0390d08ce3a" name="Default Changelist" comment="" />
<ignored path="$PROJECT_DIR$/target/" />
<ignored path="$PROJECT_DIR$/FlumeDynamicInterceptor/target/" />
<option name="EXCLUDED_CONVERTED_TO_IGNORED" value="true" />
<option name="SHOW_DIALOG" value="false" />
<option name="HIGHLIGHT_CONFLICTS" value="true" />
<option name="HIGHLIGHT_NON_ACTIVE_CHANGELIST" value="false" />
<option name="LAST_RESOLUTION" value="IGNORE" />
</component>
<component name="FUSProjectUsageTrigger">
<session id="-2092497702">
<usages-collector id="statistics.lifecycle.project">
<counts>
<entry key="project.closed" value="1" />
<entry key="project.open.time.1" value="1" />
<entry key="project.opened" value="1" />
</counts>
</usages-collector>
<usages-collector id="statistics.file.extensions.open">
<counts>
<entry key="java" value="8" />
<entry key="properties" value="1" />
</counts>
</usages-collector>
<usages-collector id="statistics.file.types.open">
<counts>
<entry key="JAVA" value="8" />
<entry key="Properties" value="1" />
</counts>
</usages-collector>
<usages-collector id="statistics.file.extensions.edit">
<counts>
<entry key="java" value="32" />
</counts>
</usages-collector>
<usages-collector id="statistics.file.types.edit">
<counts>
<entry key="JAVA" value="32" />
</counts>
</usages-collector>
</session>
</component>
<component name="FileEditorManager">
<leaf>
<file pinned="false" current-in-tab="false">
<entry file="file://$PROJECT_DIR$/FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/FlumeDynamicApp.java">
<provider selected="true" editor-type-id="text-editor">
<state relative-caret-position="-968">
<caret line="270" column="48" selection-start-line="270" selection-start-column="48" selection-end-line="270" selection-end-column="48" />
</state>
</provider>
</entry>
</file>
<file pinned="false" current-in-tab="false">
<entry file="file://$PROJECT_DIR$/FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/utils/hbase/HBaseUtils.java">
<provider selected="true" editor-type-id="text-editor">
<state relative-caret-position="-1819">
<caret line="78" column="27" selection-start-line="78" selection-start-column="27" selection-end-line="78" selection-end-column="27" />
</state>
</provider>
</entry>
</file>
<file pinned="false" current-in-tab="false">
<entry file="file://$PROJECT_DIR$/FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/utils/http/HttpClientUtil.java">
<provider selected="true" editor-type-id="text-editor">
<state relative-caret-position="665">
<caret line="52" selection-start-line="52" selection-end-line="52" />
<folding>
<element signature="e#1795#1841#0" />
</folding>
</state>
</provider>
</entry>
</file>
<file pinned="false" current-in-tab="false">
<entry file="file://$PROJECT_DIR$/FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/utils/json/JsonParseUtil.java">
<provider selected="true" editor-type-id="text-editor">
<state relative-caret-position="435">
<caret line="193" column="28" selection-start-line="193" selection-start-column="28" selection-end-line="193" selection-end-column="28" />
</state>
</provider>
</entry>
</file>
<file pinned="false" current-in-tab="false">
<entry file="file://$PROJECT_DIR$/FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/utils/system/FlowWriteConfigurations.java">
<provider selected="true" editor-type-id="text-editor">
<state relative-caret-position="391">
<caret line="17" column="33" lean-forward="true" selection-start-line="17" selection-start-column="33" selection-end-line="17" selection-end-column="33" />
</state>
</provider>
</entry>
</file>
<file pinned="false" current-in-tab="false">
<entry file="file://$PROJECT_DIR$/FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/common/FlowWriteConfig.java">
<provider selected="true" editor-type-id="text-editor">
<state relative-caret-position="481">
<caret line="42" column="93" lean-forward="true" selection-start-line="42" selection-start-column="93" selection-end-line="42" selection-end-column="93" />
</state>
</provider>
</entry>
</file>
<file pinned="false" current-in-tab="true">
<entry file="file://$PROJECT_DIR$/FlumeDynamicInterceptor/properties/service_flow_config.properties">
<provider selected="true" editor-type-id="text-editor">
<state relative-caret-position="276">
<caret line="12" selection-start-line="12" selection-end-line="12" />
</state>
</provider>
</entry>
</file>
</leaf>
</component>
<component name="IdeDocumentHistory">
<option name="CHANGED_PATHS">
<list>
<option value="$PROJECT_DIR$/FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/FlumeDynamicApp.java" />
<option value="$PROJECT_DIR$/FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/utils/hbase/HBaseUtils.java" />
<option value="$PROJECT_DIR$/FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/utils/http/HttpClientUtil.java" />
<option value="$PROJECT_DIR$/FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/utils/json/JsonParseUtil.java" />
<option value="$PROJECT_DIR$/FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/utils/system/FlowWriteConfigurations.java" />
</list>
</option>
</component>
<component name="JsBuildToolGruntFileManager" detection-done="true" sorting="DEFINITION_ORDER" />
<component name="JsBuildToolPackageJson" detection-done="true" sorting="DEFINITION_ORDER" />
<component name="JsGulpfileManager">
<detection-done>true</detection-done>
<sorting>DEFINITION_ORDER</sorting>
</component>
<component name="MavenImportPreferences">
<option name="generalSettings">
<MavenGeneralSettings>
<option name="localRepository" value="E:\dev_code\repository_maven" />
<option name="mavenHome" value="$PROJECT_DIR$/../../../../../../apache-maven-3.2.1" />
<option name="userSettingsFile" value="E:\dev_code\apache-maven-3.2.1\conf\settings.xml" />
</MavenGeneralSettings>
</option>
</component>
<component name="ProjectFrameBounds" extendedState="6">
<option name="x" value="850" />
<option name="y" value="92" />
<option name="width" value="950" />
<option name="height" value="846" />
</component>
<component name="ProjectView">
<navigator proportions="" version="1">
<foldersAlwaysOnTop value="true" />
</navigator>
<panes>
<pane id="Scope" />
<pane id="ProjectPane">
<subPane>
<expand>
<path>
<item name="dynamic_complement" type="b2602c69:ProjectViewProjectNode" />
<item name="dynamic_complement" type="462c0819:PsiDirectoryNode" />
</path>
<path>
<item name="dynamic_complement" type="b2602c69:ProjectViewProjectNode" />
<item name="dynamic_complement" type="462c0819:PsiDirectoryNode" />
<item name="FlumeDynamicInterceptor" type="462c0819:PsiDirectoryNode" />
</path>
<path>
<item name="dynamic_complement" type="b2602c69:ProjectViewProjectNode" />
<item name="dynamic_complement" type="462c0819:PsiDirectoryNode" />
<item name="FlumeDynamicInterceptor" type="462c0819:PsiDirectoryNode" />
<item name="properties" type="462c0819:PsiDirectoryNode" />
</path>
<path>
<item name="dynamic_complement" type="b2602c69:ProjectViewProjectNode" />
<item name="dynamic_complement" type="462c0819:PsiDirectoryNode" />
<item name="FlumeDynamicInterceptor" type="462c0819:PsiDirectoryNode" />
<item name="src" type="462c0819:PsiDirectoryNode" />
</path>
<path>
<item name="dynamic_complement" type="b2602c69:ProjectViewProjectNode" />
<item name="dynamic_complement" type="462c0819:PsiDirectoryNode" />
<item name="FlumeDynamicInterceptor" type="462c0819:PsiDirectoryNode" />
<item name="src" type="462c0819:PsiDirectoryNode" />
<item name="main" type="462c0819:PsiDirectoryNode" />
</path>
<path>
<item name="dynamic_complement" type="b2602c69:ProjectViewProjectNode" />
<item name="dynamic_complement" type="462c0819:PsiDirectoryNode" />
<item name="FlumeDynamicInterceptor" type="462c0819:PsiDirectoryNode" />
<item name="src" type="462c0819:PsiDirectoryNode" />
<item name="main" type="462c0819:PsiDirectoryNode" />
<item name="java" type="462c0819:PsiDirectoryNode" />
</path>
<path>
<item name="dynamic_complement" type="b2602c69:ProjectViewProjectNode" />
<item name="dynamic_complement" type="462c0819:PsiDirectoryNode" />
<item name="FlumeDynamicInterceptor" type="462c0819:PsiDirectoryNode" />
<item name="src" type="462c0819:PsiDirectoryNode" />
<item name="main" type="462c0819:PsiDirectoryNode" />
<item name="java" type="462c0819:PsiDirectoryNode" />
<item name="interceptor" type="462c0819:PsiDirectoryNode" />
</path>
<path>
<item name="dynamic_complement" type="b2602c69:ProjectViewProjectNode" />
<item name="dynamic_complement" type="462c0819:PsiDirectoryNode" />
<item name="FlumeDynamicInterceptor" type="462c0819:PsiDirectoryNode" />
<item name="src" type="462c0819:PsiDirectoryNode" />
<item name="main" type="462c0819:PsiDirectoryNode" />
<item name="java" type="462c0819:PsiDirectoryNode" />
<item name="interceptor" type="462c0819:PsiDirectoryNode" />
<item name="common" type="462c0819:PsiDirectoryNode" />
</path>
<path>
<item name="dynamic_complement" type="b2602c69:ProjectViewProjectNode" />
<item name="dynamic_complement" type="462c0819:PsiDirectoryNode" />
<item name="FlumeDynamicInterceptor" type="462c0819:PsiDirectoryNode" />
<item name="src" type="462c0819:PsiDirectoryNode" />
<item name="main" type="462c0819:PsiDirectoryNode" />
<item name="java" type="462c0819:PsiDirectoryNode" />
<item name="interceptor" type="462c0819:PsiDirectoryNode" />
<item name="utils" type="462c0819:PsiDirectoryNode" />
</path>
<path>
<item name="dynamic_complement" type="b2602c69:ProjectViewProjectNode" />
<item name="dynamic_complement" type="462c0819:PsiDirectoryNode" />
<item name="FlumeDynamicInterceptor" type="462c0819:PsiDirectoryNode" />
<item name="src" type="462c0819:PsiDirectoryNode" />
<item name="main" type="462c0819:PsiDirectoryNode" />
<item name="java" type="462c0819:PsiDirectoryNode" />
<item name="interceptor" type="462c0819:PsiDirectoryNode" />
<item name="utils" type="462c0819:PsiDirectoryNode" />
<item name="hbase" type="462c0819:PsiDirectoryNode" />
</path>
<path>
<item name="dynamic_complement" type="b2602c69:ProjectViewProjectNode" />
<item name="dynamic_complement" type="462c0819:PsiDirectoryNode" />
<item name="FlumeDynamicInterceptor" type="462c0819:PsiDirectoryNode" />
<item name="src" type="462c0819:PsiDirectoryNode" />
<item name="main" type="462c0819:PsiDirectoryNode" />
<item name="java" type="462c0819:PsiDirectoryNode" />
<item name="interceptor" type="462c0819:PsiDirectoryNode" />
<item name="utils" type="462c0819:PsiDirectoryNode" />
<item name="http" type="462c0819:PsiDirectoryNode" />
</path>
<path>
<item name="dynamic_complement" type="b2602c69:ProjectViewProjectNode" />
<item name="dynamic_complement" type="462c0819:PsiDirectoryNode" />
<item name="FlumeDynamicInterceptor" type="462c0819:PsiDirectoryNode" />
<item name="src" type="462c0819:PsiDirectoryNode" />
<item name="main" type="462c0819:PsiDirectoryNode" />
<item name="java" type="462c0819:PsiDirectoryNode" />
<item name="interceptor" type="462c0819:PsiDirectoryNode" />
<item name="utils" type="462c0819:PsiDirectoryNode" />
<item name="json" type="462c0819:PsiDirectoryNode" />
</path>
<path>
<item name="dynamic_complement" type="b2602c69:ProjectViewProjectNode" />
<item name="dynamic_complement" type="462c0819:PsiDirectoryNode" />
<item name="FlumeDynamicInterceptor" type="462c0819:PsiDirectoryNode" />
<item name="src" type="462c0819:PsiDirectoryNode" />
<item name="main" type="462c0819:PsiDirectoryNode" />
<item name="java" type="462c0819:PsiDirectoryNode" />
<item name="interceptor" type="462c0819:PsiDirectoryNode" />
<item name="utils" type="462c0819:PsiDirectoryNode" />
<item name="system" type="462c0819:PsiDirectoryNode" />
</path>
<path>
<item name="dynamic_complement" type="b2602c69:ProjectViewProjectNode" />
<item name="dynamic_complement" type="462c0819:PsiDirectoryNode" />
<item name="FlumeDynamicInterceptor" type="462c0819:PsiDirectoryNode" />
<item name="src" type="462c0819:PsiDirectoryNode" />
<item name="test" type="462c0819:PsiDirectoryNode" />
</path>
<path>
<item name="dynamic_complement" type="b2602c69:ProjectViewProjectNode" />
<item name="dynamic_complement" type="462c0819:PsiDirectoryNode" />
<item name="src" type="462c0819:PsiDirectoryNode" />
</path>
<path>
<item name="dynamic_complement" type="b2602c69:ProjectViewProjectNode" />
<item name="dynamic_complement" type="462c0819:PsiDirectoryNode" />
<item name="src" type="462c0819:PsiDirectoryNode" />
<item name="main" type="462c0819:PsiDirectoryNode" />
</path>
<path>
<item name="dynamic_complement" type="b2602c69:ProjectViewProjectNode" />
<item name="dynamic_complement" type="462c0819:PsiDirectoryNode" />
<item name="src" type="462c0819:PsiDirectoryNode" />
<item name="test" type="462c0819:PsiDirectoryNode" />
</path>
</expand>
<select />
</subPane>
</pane>
<pane id="PackagesPane" />
</panes>
</component>
<component name="PropertiesComponent">
<property name="WebServerToolWindowFactoryState" value="false" />
<property name="aspect.path.notification.shown" value="true" />
<property name="com.android.tools.idea.instantapp.provision.ProvisionBeforeRunTaskProvider.myTimeStamp" value="1591960947841" />
<property name="go.gopath.indexing.explicitly.defined" value="true" />
<property name="nodejs_interpreter_path.stuck_in_default_project" value="undefined stuck path" />
<property name="nodejs_npm_path_reset_for_default_project" value="true" />
</component>
<component name="RunDashboard">
<option name="ruleStates">
<list>
<RuleState>
<option name="name" value="ConfigurationTypeDashboardGroupingRule" />
</RuleState>
<RuleState>
<option name="name" value="StatusDashboardGroupingRule" />
</RuleState>
</list>
</option>
</component>
<component name="SvnConfiguration">
<configuration />
</component>
<component name="TaskManager">
<task active="true" id="Default" summary="Default task">
<changelist id="7803fbe0-0492-4e54-9d86-e0390d08ce3a" name="Default Changelist" comment="" />
<created>1591958573741</created>
<option name="number" value="Default" />
<option name="presentableId" value="Default" />
<updated>1591958573741</updated>
<workItem from="1591958576333" duration="1762000" />
</task>
<servers />
</component>
<component name="TimeTrackingManager">
<option name="totallyTimeSpent" value="1762000" />
</component>
<component name="ToolWindowManager">
<frame x="-8" y="-8" width="1936" height="1035" extended-state="6" />
<layout>
<window_info id="Image Layers" />
<window_info id="Designer" />
<window_info id="UI Designer" />
<window_info id="Capture Tool" />
<window_info id="Favorites" side_tool="true" />
<window_info active="true" content_ui="combo" id="Project" order="0" visible="true" weight="0.25" />
<window_info id="Structure" order="1" side_tool="true" weight="0.25" />
<window_info anchor="bottom" id="Java Enterprise" />
<window_info anchor="bottom" id="Database Changes" show_stripe_button="false" />
<window_info anchor="bottom" id="Version Control" show_stripe_button="false" />
<window_info anchor="bottom" id="Terminal" />
<window_info anchor="bottom" id="Event Log" side_tool="true" />
<window_info anchor="bottom" id="Message" order="0" />
<window_info anchor="bottom" id="Find" order="1" />
<window_info anchor="bottom" id="Run" order="2" />
<window_info anchor="bottom" id="Debug" order="3" weight="0.4" />
<window_info anchor="bottom" id="Cvs" order="4" weight="0.25" />
<window_info anchor="bottom" id="Inspection" order="5" weight="0.4" />
<window_info anchor="bottom" id="TODO" order="6" />
<window_info anchor="right" id="Palette" />
<window_info anchor="right" id="Theme Preview" />
<window_info anchor="right" id="Bean Validation" />
<window_info anchor="right" id="Capture Analysis" />
<window_info anchor="right" id="Palette&#9;" />
<window_info anchor="right" id="Database" />
<window_info anchor="right" id="Maven Projects" />
<window_info anchor="right" id="Commander" internal_type="SLIDING" order="0" type="SLIDING" weight="0.4" />
<window_info anchor="right" id="Ant Build" order="1" weight="0.25" />
<window_info anchor="right" content_ui="combo" id="Hierarchy" order="2" weight="0.25" />
</layout>
</component>
<component name="TypeScriptGeneratedFilesManager">
<option name="version" value="1" />
</component>
<component name="VcsContentAnnotationSettings">
<option name="myLimit" value="2678400000" />
</component>
<component name="editorHistoryManager">
<entry file="file://$PROJECT_DIR$/FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/utils/hbase/HBaseUtils.java">
<provider selected="true" editor-type-id="text-editor">
<state relative-caret-position="-1819">
<caret line="78" column="27" selection-start-line="78" selection-start-column="27" selection-end-line="78" selection-end-column="27" />
</state>
</provider>
</entry>
<entry file="file://$PROJECT_DIR$/FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/utils/json/BAKJsonParseUtil.java">
<provider selected="true" editor-type-id="text-editor">
<state relative-caret-position="-4073">
<folding>
<element signature="e#0#6658#0" expanded="true" />
</folding>
</state>
</provider>
</entry>
<entry file="file://$PROJECT_DIR$/FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/utils/http/HttpClientUtil.java">
<provider selected="true" editor-type-id="text-editor">
<state relative-caret-position="665">
<caret line="52" selection-start-line="52" selection-end-line="52" />
<folding>
<element signature="e#1795#1841#0" />
</folding>
</state>
</provider>
</entry>
<entry file="file://$PROJECT_DIR$/FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/utils/json/JsonParseUtil.java">
<provider selected="true" editor-type-id="text-editor">
<state relative-caret-position="435">
<caret line="193" column="28" selection-start-line="193" selection-start-column="28" selection-end-line="193" selection-end-column="28" />
</state>
</provider>
</entry>
<entry file="file://$PROJECT_DIR$/FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/utils/system/SnowflakeId.java">
<provider selected="true" editor-type-id="text-editor">
<state>
<folding>
<element signature="e#0#7712#0" expanded="true" />
</folding>
</state>
</provider>
</entry>
<entry file="file://$PROJECT_DIR$/FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/FlumeDynamicApp.java">
<provider selected="true" editor-type-id="text-editor">
<state relative-caret-position="-968">
<caret line="270" column="48" selection-start-line="270" selection-start-column="48" selection-end-line="270" selection-end-column="48" />
</state>
</provider>
</entry>
<entry file="file://$PROJECT_DIR$/FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/utils/system/FlowWriteConfigurations.java">
<provider selected="true" editor-type-id="text-editor">
<state relative-caret-position="391">
<caret line="17" column="33" lean-forward="true" selection-start-line="17" selection-start-column="33" selection-end-line="17" selection-end-column="33" />
</state>
</provider>
</entry>
<entry file="file://$PROJECT_DIR$/FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/common/FlowWriteConfig.java">
<provider selected="true" editor-type-id="text-editor">
<state relative-caret-position="481">
<caret line="42" column="93" lean-forward="true" selection-start-line="42" selection-start-column="93" selection-end-line="42" selection-end-column="93" />
</state>
</provider>
</entry>
<entry file="file://$PROJECT_DIR$/FlumeDynamicInterceptor/properties/service_flow_config.properties">
<provider selected="true" editor-type-id="text-editor">
<state relative-caret-position="276">
<caret line="12" selection-start-line="12" selection-end-line="12" />
</state>
</provider>
</entry>
</component>
<component name="masterDetails">
<states>
<state key="ProjectJDKs.UI">
<settings>
<last-edited>1.8</last-edited>
<splitter-proportions>
<option name="proportions">
<list>
<option value="0.2" />
</list>
</option>
</splitter-proportions>
</settings>
</state>
</states>
</component>
</project>

294
FlumeDynamicInterceptor/pom.xml Normal file
View File

@@ -0,0 +1,294 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>dynamic_complement</artifactId>
<groupId>com.zdjizhi</groupId>
<version>1.0</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>FlumeDynamicInterceptor</artifactId>
<repositories>
<repository>
<id>nexus</id>
<name>Team Nexus Repository</name>
<url>http://192.168.40.125:8099/content/groups/public</url>
</repository>
<repository>
<id>ebi</id>
<name>www.ebi.ac.uk</name>
<url>http://www.ebi.ac.uk/intact/maven/nexus/content/groups/public/</url>
</repository>
</repositories>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<flume.version>1.9.0</flume.version>
<hbase.version>2.2.1</hbase.version>
<!--<hadoop.version>2.8.5</hadoop.version>-->
<!--<hbase.version>1.4.9</hbase.version>-->
<!--<hadoop.version>2.7.1</hadoop.version>-->
</properties>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>2.4.1</version>
<configuration>
<createDependencyReducedPom>true</createDependencyReducedPom>
</configuration>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<transformers>
<transformer
implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
<transformer
implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>com.zdjizhi.flume.interceptor.FlumeDynamicApp</mainClass>
</transformer>
<transformer
implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>META-INF/spring.handlers</resource>
</transformer>
<transformer
implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>META-INF/spring.schemas</resource>
</transformer>
</transformers>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>exec-maven-plugin</artifactId>
<version>1.2.1</version>
<executions>
<execution>
<goals>
<goal>exec</goal>
</goals>
</execution>
</executions>
<configuration>
<executable>java</executable>
<includeProjectDependencies>true</includeProjectDependencies>
<includePluginDependencies>false</includePluginDependencies>
<classpathScope>compile</classpathScope>
<mainClass>com.zdjizhi.flume.interceptor.FlumeDynamicApp</mainClass>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>2.3.2</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
</configuration>
</plugin>
</plugins>
<resources>
<resource>
<directory>properties</directory>
<includes>
<include>**/*.properties</include>
<include>**/*.xml</include>
</includes>
<filtering>false</filtering>
</resource>
<!--<resource>-->
<!--<directory>src/main/java</directory>-->
<!--<includes>-->
<!--<include>log4j.properties</include>-->
<!--</includes>-->
<!--<filtering>false</filtering>-->
<!--</resource>-->
</resources>
</build>
<dependencies>
<dependency>
<groupId>org.apache.flume</groupId>
<artifactId>flume-ng-core</artifactId>
<version>${flume.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>com.zdjizhi</groupId>
<artifactId>galaxy</artifactId>
<!-- <version>1.0.2</version>-->
<version>1.0.3</version>
<exclusions>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
<exclusion>
<artifactId>log4j-over-slf4j</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
<exclusion>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.47</version>
</dependency>
<!--<dependency>-->
<!--<groupId>redis.clients</groupId>-->
<!--<artifactId>jedis</artifactId>-->
<!--<version>2.8.1</version>-->
<!--</dependency>-->
<!-- <dependency>-->
<!-- <groupId>org.apache.zookeeper</groupId>-->
<!-- <artifactId>zookeeper</artifactId>-->
<!-- <version>3.4.9</version>-->
<!-- <exclusions>-->
<!-- <exclusion>-->
<!-- <artifactId>slf4j-log4j12</artifactId>-->
<!-- <groupId>org.slf4j</groupId>-->
<!-- </exclusion>-->
<!-- <exclusion>-->
<!-- <artifactId>log4j-over-slf4j</artifactId>-->
<!-- <groupId>org.slf4j</groupId>-->
<!-- </exclusion>-->
<!-- </exclusions>-->
<!-- </dependency>-->
<dependency>
<groupId>cglib</groupId>
<artifactId>cglib-nodep</artifactId>
<version>3.2.4</version>
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-client</artifactId>
<version>${hbase.version}</version>
<exclusions>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
<exclusion>
<artifactId>log4j-over-slf4j</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-server</artifactId>
<version>${hbase.version}</version>
<exclusions>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
<exclusion>
<artifactId>log4j-over-slf4j</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
</dependency>
<!--<dependency>-->
<!--<groupId>org.apache.hadoop</groupId>-->
<!--<artifactId>hadoop-common</artifactId>-->
<!--<version>${hadoop.version}</version>-->
<!--<exclusions>-->
<!--<exclusion>-->
<!--<artifactId>slf4j-log4j12</artifactId>-->
<!--<groupId>org.slf4j</groupId>-->
<!--</exclusion>-->
<!--<exclusion>-->
<!--<artifactId>log4j-over-slf4j</artifactId>-->
<!--<groupId>org.slf4j</groupId>-->
<!--</exclusion>-->
<!--</exclusions>-->
<!--</dependency>-->
<!--<dependency>-->
<!--<groupId>org.apache.hadoop</groupId>-->
<!--<artifactId>hadoop-client</artifactId>-->
<!--<version>${hadoop.version}</version>-->
<!--<exclusions>-->
<!--<exclusion>-->
<!--<artifactId>slf4j-log4j12</artifactId>-->
<!--<groupId>org.slf4j</groupId>-->
<!--</exclusion>-->
<!--<exclusion>-->
<!--<artifactId>log4j-over-slf4j</artifactId>-->
<!--<groupId>org.slf4j</groupId>-->
<!--</exclusion>-->
<!--</exclusions>-->
<!--</dependency>-->
<!--<dependency>-->
<!--<groupId>org.apache.hadoop</groupId>-->
<!--<artifactId>hadoop-hdfs</artifactId>-->
<!--<version>${hadoop.version}</version>-->
<!--<exclusions>-->
<!--<exclusion>-->
<!--<artifactId>slf4j-log4j12</artifactId>-->
<!--<groupId>org.slf4j</groupId>-->
<!--</exclusion>-->
<!--<exclusion>-->
<!--<artifactId>log4j-over-slf4j</artifactId>-->
<!--<groupId>org.slf4j</groupId>-->
<!--</exclusion>-->
<!--</exclusions>-->
<!--</dependency>-->
<!-- <dependency>-->
<!-- <groupId>org.junit.jupiter</groupId>-->
<!-- <artifactId>junit-jupiter-api</artifactId>-->
<!-- <version>5.3.2</version>-->
<!-- <scope>compile</scope>-->
<!-- </dependency>-->
<!--<dependency>-->
<!--<groupId>com.google.guava</groupId>-->
<!--<artifactId>guava</artifactId>-->
<!--&lt;!&ndash;<version>18.0</version>&ndash;&gt;-->
<!--<version>11.0.2</version>-->
<!--</dependency>-->
<!--<dependency>-->
<!--<groupId>org.apache.httpcomponents</groupId>-->
<!--<artifactId>httpclient</artifactId>-->
<!--<version>4.5.2</version>-->
<!--</dependency>-->
<!--<dependency>-->
<!--<groupId>org.apache.httpcomponents</groupId>-->
<!--<artifactId>httpcore</artifactId>-->
<!--<version>4.4.1</version>-->
<!--</dependency>-->
</dependencies>
</project>

23
FlumeDynamicInterceptor/properties/service_flow_config.properties Normal file
View File

@@ -0,0 +1,23 @@
#kafka broker下的topic名称
#kafka.topic=SESSION-TEST-LOG
#数据中心（UID）
#data.center.id.num=15
#zookeeper.servers=192.168.40.207:2181
#用于过滤对准用户名
#check.ip.scope=10,100,192
#获取Schema的网址
schema.http=http://192.168.40.224:9999/metadata/schema/v1/fields/security_event_log
#定位库位置
ip.library=E:/workdata/ip_all/
#hbase-zookeeper地址
#hbase.zookeeper.servers=192.168.40.186:2182
hbase.zookeeper.servers=192.168.40.224:2181
#hbase表名
hbase.table.name=subscriber_info
#邮件默认编码
mail.default.charset=UTF-8

343
FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/FlumeDynamicApp.java Normal file
View File

@@ -0,0 +1,343 @@
package com.zdjizhi.flume.interceptor;
import com.alibaba.fastjson.JSONObject;
import com.google.common.base.Preconditions;
import com.zdjizhi.flume.interceptor.common.FlowWriteConfig;
import com.zdjizhi.flume.interceptor.utils.hbase.HBaseUtils;
import com.zdjizhi.flume.interceptor.utils.json.JsonParseUtil;
import com.zdjizhi.utils.Encodes;
import com.zdjizhi.utils.FormatUtils;
import com.zdjizhi.utils.IpLookup;
import com.zdjizhi.utils.StringUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import org.apache.log4j.Logger;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
public class FlumeDynamicApp implements Interceptor {
private static Logger logger = Logger.getLogger(FlumeDynamicApp.class);
private static IpLookup ipLookup;
private static FormatUtils formatUtils;
private String schemaHttpUrl;
private String uidZookeeperIp;
private long dataCenterIdNum;
private String ipDatPath;
private String hbaseZookeeperIp;
private String hbaseTableName;
private static HashMap<String, Class> map;
private static Object mapObject;
private static ArrayList<String[]> jobList;
public void initialize() {
map = JsonParseUtil.getMapFromhttp(schemaHttpUrl);
mapObject = JsonParseUtil.generateObject(map);
jobList = JsonParseUtil.getJobListFromHttp(schemaHttpUrl);
//载入工具类
formatUtils = new FormatUtils.Builder(false).build();
//载入定位库
ipLookup = new IpLookup.Builder(false)
/**
* v1.0.2-com.zdjizhi.galaxy
*/
// .loadDataFileV4(ipDatPath + "Kazakhstan.mmdb")
// .loadDataFileV6(ipDatPath + "Kazakhstan.mmdb")
// .loadAsnDataFileV4(ipDatPath + "asn_v4.mmdb")
// .loadAsnDataFileV6(ipDatPath + "asn_v6.mmdb")
/**
* v1.0.3-com.zdjizhi.galaxy
*/
.loadDataFile(ipDatPath + "all_ip_info_v4.mmdb")
.loadDataFilePrivateV4(ipDatPath + "ip_private.mmdb")
.loadDataFilePrivateV6(ipDatPath + "ip_private.mmdb")
.loadAsnDataFile(ipDatPath + "asn_v4.mmdb")
.loadAsnDataFileV6(ipDatPath + "asn_v6.mmdb")
.build();
}
public FlumeDynamicApp(String schemaHttpUrl, String zookeeperIp, long dataCenterIdNum, String ipDatPath, String hbaseZookeeperIp, String hbaseTableName) {
this.schemaHttpUrl = schemaHttpUrl;
this.uidZookeeperIp = zookeeperIp;
this.dataCenterIdNum = dataCenterIdNum;
this.ipDatPath = ipDatPath;
this.hbaseZookeeperIp = hbaseZookeeperIp;
this.hbaseTableName = hbaseTableName;
}
public Event intercept(Event event) {
String message = null;
try {
message = new String(event.getBody(), "utf-8");
} catch (UnsupportedEncodingException e) {
message = new String(event.getBody());
}
try {
if (StringUtils.isNotBlank(message)) {
message = dealCommonMessage(message);
event.setBody(message.getBytes());
return event;
}
} catch (Exception e) {
logger.error("FlumeDynamicApp intercept(Event event) method is error===>{" + e + "}<===");
e.printStackTrace();
}
return event;
}
public List<Event> intercept(List<Event> list) {
List resultList = new ArrayList();
for (Event event : list) {
Event r = intercept(event);
if (r != null) {
resultList.add(r);
}
}
return resultList;
}
public void close() {
logger.warn("FlumeDynamicApp is closed.");
}
/**
* 解析日志，并补全
* 补domain,补subscriber_id
*
* @param message Security原始日志
* @return 补全后的日志
* <p>
*/
private String dealCommonMessage(String message) {
Object object = JSONObject.parseObject(message, mapObject.getClass());
try {
for (String[] strings : jobList) {
Object name = JsonParseUtil.getValue(object, strings[0]);
String appendToKeyName = strings[1];
String functionName = strings[2];
Object param = null;
if (strings[3] != null) {
param = JsonParseUtil.getValue(object, strings[3]);
}
switch (functionName) {
case "current_timestamp":
JsonParseUtil.setValue(object, appendToKeyName, getCurrentTime());
break;
case "snowflake_id":
JsonParseUtil.setValue(object, appendToKeyName, formatUtils.getSnowflakeId(uidZookeeperIp, dataCenterIdNum));
break;
case "geo_ip_detail":
if (name != null) {
JsonParseUtil.setValue(object, appendToKeyName, getGeoIpDetail(name.toString()));
}
break;
case "geo_asn":
if (name != null) {
JsonParseUtil.setValue(object, appendToKeyName, getGeoAsn(name.toString()));
}
break;
case "get_value":
if (name != null) {
JsonParseUtil.setValue(object, appendToKeyName, name.toString());
}
break;
case "radius_match":
if (name != null) {
JsonParseUtil.setValue(object, appendToKeyName, radiusMatch(name.toString(), hbaseZookeeperIp, hbaseTableName));
}
break;
case "geo_ip_country":
if (name != null) {
JsonParseUtil.setValue(object, appendToKeyName, getGeoIpCountry(name.toString()));
}
break;
case "decode_of_base64":
if (name != null) {
if (param != null) {
JsonParseUtil.setValue(object, appendToKeyName, Encodes.decodeBase64String(name.toString(), param.toString()));
} else {
JsonParseUtil.setValue(object, appendToKeyName, Encodes.decodeBase64String(name.toString(), FlowWriteConfig.MAIL_DEFAULT_CHARSET));
}
}
break;
case "sub_domain":
if (name != null) {
Object appendTo = JsonParseUtil.getValue(object, appendToKeyName);
if (appendTo == null || StringUtil.isBlank(appendTo.toString())) {
JsonParseUtil.setValue(object, appendToKeyName, replaceGetTopDomain(FormatUtils.getTopPrivateDomain(name.toString())));
}
}
break;
default:
}
}
return JSONObject.toJSONString(object);
} catch (Exception e) {
logger.error("FlumeDynamicApp dealCommonMessage is error===>{" + e + "}<===");
e.printStackTrace();
// return "";
return message;//返回原数据
}
}
/**
* 有host根据host获取域名,有sni通过sni获取域名
* 这里是直接根据传入的获取域名
*
* @param url
* @return 顶级域名
*/
private String replaceGetTopDomain(String url) {
return url.replaceAll("InternetDomainName\\{name=", "").replaceAll("\\}", "");
}
/**
* 生成当前时间戳的操作
*/
private int getCurrentTime() {
return (int) (System.currentTimeMillis() / 1000);
}
/**
* 根据clientIp获取location信息
*
* @param ip
* @return
*/
private String getGeoIpDetail(String ip) {
return ipLookup.cityLookupDetail(ip);
}
/**
* 根据ip获取asn信息
*
* @param ip
* @return
*/
private String getGeoAsn(String ip) {
// return ipLookup.asnLookup(ip, true);//v1.0.2-com.zdjizhi.galaxy
return ipLookup.asnLookup(ip);//v1.0.3-com.zdjizhi.galaxy
}
/**
* 根据ip获取country信息
*
* @param ip
* @return
*/
private String getGeoIpCountry(String ip) {
return ipLookup.countryLookup(ip);
}
/**
* radius借助hbase补齐
*
* @param ip
* @return
*/
private static String radiusMatch(String ip, String hbaseZookeeper, String hbaseTable) {
return HBaseUtils.getAccount(ip, hbaseZookeeper, hbaseTable);
}
public static class FlumeDynamicAppBuilder implements Interceptor.Builder {
private String schemaHttpUrl;
private String uidZookeeperIp;
private long dataCenterIdNum;
private String ipDatPath;
private String hbaseZookeeperIp;
private String hbaseTableName;
public Interceptor build() {
return new FlumeDynamicApp(this.schemaHttpUrl,
this.uidZookeeperIp, this.dataCenterIdNum,
this.ipDatPath,
this.hbaseZookeeperIp, this.hbaseTableName);
}
public void configure(Context context) {
try {
this.schemaHttpUrl = context.getString("schemaHttpUrl", "");
Preconditions.checkNotNull("".equals(schemaHttpUrl), "schemaHttpUrl must be set!!");
logger.info("FlumeDynamicApp Read schemaHttpUrl from configuration : " + schemaHttpUrl);
} catch (IllegalArgumentException e) {
throw new IllegalArgumentException("FlumeDynamicApp schemaHttpUrl invalid", e);
} catch (Exception e) {
logger.error("FlumeDynamicApp Get schemaHttpUrl is error : " + e);
}
try {
this.uidZookeeperIp = context.getString("uidZookeeperIp", "");
Preconditions.checkNotNull("".equals(uidZookeeperIp), "uidZookeeperIp must be set!!");
logger.info("FlumeDynamicApp Read uidZookeeperIp from configuration : " + uidZookeeperIp);
} catch (IllegalArgumentException e) {
throw new IllegalArgumentException("FlumeDynamicApp uidZookeeperIp invalid", e);
} catch (Exception e) {
logger.error("FlumeDynamicApp Get uidZookeeperIp is error : " + e);
}
try {
this.dataCenterIdNum = context.getInteger("dataCenterIdNum", RandomUtils.nextInt(0, 16));//0-16,不包含16,即只有0-15
if (dataCenterIdNum < 0L || dataCenterIdNum > 15L) {
logger.error("\ndataCenterIdNum is out of effective range,dataCenterIdNum must be greater than 0(include 0) and less than 16(exclusive 16),but there is " + dataCenterIdNum);
this.dataCenterIdNum = RandomUtils.nextInt(0, 16);
logger.error("\nso system will auto get a random num between 0 and 15,so this random dataCenterIdNum is " + dataCenterIdNum);
} else {
logger.info("FlumeDynamicApp Read dataCenterIdNum from configuration : " + dataCenterIdNum);
}
} catch (IllegalArgumentException e) {
throw new IllegalArgumentException("FlumeDynamicApp dataCenterIdNum invalid", e);
} catch (Exception e) {
logger.error("FlumeDynamicApp Get dataCenterIdNum is error : " + e);
}
try {
this.ipDatPath = context.getString("ipDatPath", "");
Preconditions.checkNotNull("".equals(ipDatPath), "ipDatPath must be set!!");
if ((!"".equals(ipDatPath)) && (!ipDatPath.endsWith("/"))) {
this.ipDatPath = ipDatPath + "/";
}
logger.info("FlumeDynamicApp Read ipDatPath from configuration : " + ipDatPath);
} catch (IllegalArgumentException e) {
throw new IllegalArgumentException("FlumeDynamicApp ipDatPath invalid", e);
} catch (Exception e) {
logger.error("FlumeDynamicApp Get ipDatPath is error : " + e);
}
try {
this.hbaseZookeeperIp = context.getString("hbaseZookeeperIp", "");
Preconditions.checkNotNull("".equals(hbaseZookeeperIp), "hbaseZookeeperIp must be set!!");
logger.info("FlumeDynamicApp Read hbaseZookeeperIp from configuration : " + hbaseZookeeperIp);
} catch (IllegalArgumentException e) {
throw new IllegalArgumentException("FlumeDynamicApp hbaseZookeeperIp invalid", e);
} catch (Exception e) {
logger.error("FlumeDynamicApp Get hbaseZookeeperIp is error : " + e);
}
try {
this.hbaseTableName = context.getString("hbaseTableName", "");
Preconditions.checkNotNull("".equals(hbaseTableName), "hbaseTableName must be set!!");
logger.info("FlumeDynamicApp Read hbaseTableName from configuration : " + hbaseTableName);
} catch (IllegalArgumentException e) {
throw new IllegalArgumentException("FlumeDynamicApp hbaseTableName invalid", e);
} catch (Exception e) {
logger.error("FlumeDynamicApp Get hbaseTableName is error : " + e);
}
}
}
}

53
FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/common/FlowWriteConfig.java Normal file
View File

@@ -0,0 +1,53 @@
package com.zdjizhi.flume.interceptor.common;
import com.zdjizhi.flume.interceptor.utils.system.FlowWriteConfigurations;
/**
* @author Administrator
*/
public class FlowWriteConfig {
// public static final String SEGMENTATION = ",";
/**
* System
*/
// public static final Integer DATA_CENTER_ID_NUM = FlowWriteConfigurations.getIntProperty(0, "data.center.id.num");
// public static final String CHECK_IP_SCOPE = FlowWriteConfigurations.getStringProperty(0, "check.ip.scope");
/**
* kafka
*/
// public static final String KAFKA_TOPIC = FlowWriteConfigurations.getStringProperty(0, "kafka.topic");
/**
* zookeeper
*/
// public static final String ZOOKEEPER_SERVERS = FlowWriteConfigurations.getStringProperty(0, "zookeeper.servers");
/***
* Redis
*/
// public static final String REDIS_IP = "redis.ip";
// public static final String REDIS_PORT = "redis.port";
// public static final String REDIS_TIMEOUT = "redis.timeout";
// public static final String REDIS_POOL_MAXACTIVE = "redis.pool.maxActive";
// public static final String REDIS_POOL_MAXIDLE = "redis.pool.maxIdle";
// public static final String REDIS_POOL_MAXWAIT = "redis.pool.maxWait";
// public static final String REDIS_POOL_TESTONBORROW = "redis.pool.testOnBorrow";
// public static final String REDIS_POOL_TESTONRETURN = "redis.pool.testOnReturn";
/**
* flume使用配置
*/
public static final String SCHEMA_HTTP = FlowWriteConfigurations.getStringProperty(0, "schema.http");
public static final String IP_LIBRARY = FlowWriteConfigurations.getStringProperty(0, "ip.library");
public static final String HBASE_ZOOKEEPER_SERVERS = FlowWriteConfigurations.getStringProperty(0, "hbase.zookeeper.servers");
public static final String HBASE_TABLE_NAME = FlowWriteConfigurations.getStringProperty(0, "hbase.table.name");
public static final String MAIL_DEFAULT_CHARSET = FlowWriteConfigurations.getStringProperty(0, "mail.default.charset");
}

197
FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/utils/hbase/HBaseUtils.java Normal file
View File

@@ -0,0 +1,197 @@
package com.zdjizhi.flume.interceptor.utils.hbase;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
/**
* HBase 工具类
*
* @author qidaijie
*/
public class HBaseUtils {
private final static Logger logger = Logger.getLogger(HBaseUtils.class);
private static Map<String, String> subIdMap = new HashMap<>(333334);
private static Connection connection;
private static Long time;
private static String zookeeperIp;
private static String hbaseTable;
private static HBaseUtils hBaseUtils;
private static void getHbaseInstance(String zookeeperServer, String hbaseTableName) {
hBaseUtils = new HBaseUtils(zookeeperServer, hbaseTableName);
}
/**
* 构造函数-新
*/
private HBaseUtils(String zookeeperServer, String hbaseTableName) {
zookeeperIp = zookeeperServer;
hbaseTable = hbaseTableName;
//获取连接
getHbaseConn();
//拉取所有
getAll();
//定时更新
updateHabaseCache();
}
private static void getHbaseConn() {
try {
// 管理Hbase的配置信息
Configuration configuration = HBaseConfiguration.create();
// 设置zookeeper节点
configuration.set("hbase.zookeeper.quorum", zookeeperIp);
configuration.set("hbase.client.retries.number", "3");
configuration.set("hbase.bulkload.retries.number", "3");
configuration.set("zookeeper.recovery.retry", "3");
connection = ConnectionFactory.createConnection(configuration);
time = System.currentTimeMillis();
logger.warn("HBaseUtils get HBase connection,now to getAll().");
// getAll();
} catch (IOException ioe) {
logger.error("HBaseUtils getHbaseConn() IOException===>{" + ioe + "}<===");
ioe.printStackTrace();
} catch (Exception e) {
logger.error("HBaseUtils getHbaseConn() Exception===>{" + e + "}<===");
e.printStackTrace();
}
}
/**
* 更新变量
*/
private static void change() {
long nowTime = System.currentTimeMillis();
timestampsFilter(time - 1000, nowTime + 500);
}
/**
* 获取变更内容
*
* @param startTime 开始时间
* @param endTime 结束时间
*/
private static void timestampsFilter(Long startTime, Long endTime) {
Long begin = System.currentTimeMillis();
Table table = null;
ResultScanner scanner = null;
Scan scan2 = new Scan();
try {
table = connection.getTable(TableName.valueOf("sub:" + hbaseTable));
// table = connection.getTable(TableName.valueOf("sub:" + FlowWriteConfig.HBASE_TABLE_NAME));
scan2.setTimeRange(startTime, endTime);
scanner = table.getScanner(scan2);
for (Result result : scanner) {
Cell[] cells = result.rawCells();
for (Cell cell : cells) {
String key = Bytes.toString(CellUtil.cloneRow(cell));
String value = Bytes.toString(CellUtil.cloneValue(cell));
if (subIdMap.containsKey(key)) {
if (!value.equals(subIdMap.get(key))) {
subIdMap.put(key, value);
}
} else {
subIdMap.put(key, value);
}
}
}
Long end = System.currentTimeMillis();
logger.warn("HBaseUtils Now subIdMap.keySet().size() is: " + subIdMap.keySet().size());
logger.warn("HBaseUtils Update cache timeConsuming is: " + (end - begin) + ",BeginTime: " + begin + ",EndTime: " + end);
time = endTime;
} catch (IOException ioe) {
logger.error("HBaseUtils timestampsFilter is IOException===>{" + ioe + "}<===");
ioe.printStackTrace();
} catch (Exception e) {
logger.error("HBaseUtils timestampsFilter is Exception===>{" + e + "}<===");
e.printStackTrace();
} finally {
if (scanner != null) {
scanner.close();
}
if (table != null) {
try {
table.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
/**
* 获取所有的 key value
*/
private static void getAll() {
long begin = System.currentTimeMillis();
try {
Table table = connection.getTable(TableName.valueOf("sub:" + hbaseTable));
// Table table = connection.getTable(TableName.valueOf("sub:" + FlowWriteConfig.HBASE_TABLE_NAME));
Scan scan2 = new Scan();
ResultScanner scanner = table.getScanner(scan2);
for (Result result : scanner) {
Cell[] cells = result.rawCells();
for (Cell cell : cells) {
subIdMap.put(Bytes.toString(CellUtil.cloneRow(cell)), Bytes.toString(CellUtil.cloneValue(cell)));
}
}
logger.warn("HBaseUtils Get fullAmount List size->subIdMap.size(): " + subIdMap.size());
logger.warn("HBaseUtils Get fullAmount List size->subIdMap.size() timeConsuming is: " + (System.currentTimeMillis() - begin));
scanner.close();
} catch (IOException ioe) {
logger.error("HBaseUtils getAll() is IOException===>{" + ioe + "}<===");
ioe.printStackTrace();
} catch (Exception e) {
logger.error("HBaseUtils getAll() is Exception===>{" + e + "}<===");
e.printStackTrace();
}
}
/**
* 获取 account
*
* @param clientIp client_ip
* @return account
*/
public static String getAccount(String clientIp, String hbaseZookeeper, String hbaseTable) {
if (hBaseUtils == null) {
getHbaseInstance(hbaseZookeeper, hbaseTable);
}
return subIdMap.get(clientIp);
}
/**
* 验证定时器,每隔一段时间验证一次-验证获取新的Cookie
*/
private void updateHabaseCache() {
Timer timer = new Timer();
timer.scheduleAtFixedRate(new TimerTask() {
@Override
public void run() {
try {
change();
} catch (Exception e) {
logger.error("HBaseUtils update hbaseCache is error===>{" + e + "}<===");
e.printStackTrace();
}
}
}, 1, 1000 * 60);//起始1ms,以后每隔60s
}
}

62
FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/utils/http/HttpClientUtil.java Normal file
View File

@@ -0,0 +1,62 @@
package com.zdjizhi.flume.interceptor.utils.http;
import com.alibaba.fastjson.JSON;
import org.apache.http.HttpEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
/**
* 获取网关schema工具类
*/
public class HttpClientUtil {
public static String requestByGetMethod(String s) {
CloseableHttpClient httpClient = HttpClients.createDefault();
StringBuilder entityStringBuilder = null;
try {
HttpGet get = new HttpGet(s);
CloseableHttpResponse httpResponse = null;
httpResponse = httpClient.execute(get);
try {
HttpEntity entity = httpResponse.getEntity();
entityStringBuilder = new StringBuilder();
if (null != entity) {
BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(httpResponse.getEntity().getContent(), "UTF-8"), 8 * 1024);
String line = null;
while ((line = bufferedReader.readLine()) != null) {
entityStringBuilder.append(line);
}
}
} finally {
httpResponse.close();
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
if (httpClient != null) {
httpClient.close();
}
} catch (IOException e) {
e.printStackTrace();
}
}
return entityStringBuilder.toString();
}
// public static void main(String[] args) {
//// String s = HttpClientUtil.requestByGetMethod("http://192.168.40.224:9999/metadata/schema/v1/fields/security_event_log");
//// System.out.println(s);
//// String schemaHttpRes = HttpClientUtil.requestByGetMethod("http://192.168.40.224:9999/metadata/schema/v1/fields/security_event_log");
//// String schemaHttpRes = HttpClientUtil.requestByGetMethod("http://192.168.40.151:9999/metadata/schema/v1/fields/security_event_log");
// String schemaHttpRes = HttpClientUtil.requestByGetMethod("http://192.168.40.224:9999/metadata/schema/v1/fields/connection_record_log");
// String data = JSON.parseObject(schemaHttpRes).get("data").toString();
// System.out.println(data);
// }
}

206
FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/utils/json/JsonParseUtil.java Normal file
View File

@@ -0,0 +1,206 @@
package com.zdjizhi.flume.interceptor.utils.json;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.zdjizhi.flume.interceptor.utils.http.HttpClientUtil;
import net.sf.cglib.beans.BeanGenerator;
import net.sf.cglib.beans.BeanMap;
import java.util.*;
/**
* 使用fastjson解析json的工具类
*/
public class JsonParseUtil {
/**
* 模式匹配，给定一个类型字符串返回一个类类型
*
* @param type
* @return
*/
public static Class getClassName(String type) {
Class clazz;
switch (type) {
case "int":
clazz = Integer.class;
break;
case "String":
clazz = String.class;
break;
case "long":
clazz = long.class;
break;
case "Integer":
clazz = Integer.class;
break;
case "double":
clazz = double.class;
break;
case "float":
clazz = float.class;
break;
case "char":
clazz = char.class;
break;
case "byte":
clazz = byte.class;
break;
case "boolean":
clazz = boolean.class;
break;
case "short":
clazz = short.class;
break;
default:
clazz = String.class;
}
return clazz;
}
/**
* 根据反射生成对象的方法
*
* @param properties
* @return 生成的Object类型的对象
*/
public static Object generateObject(Map properties) {
BeanGenerator generator = new BeanGenerator();
Set keySet = properties.keySet();
for (Iterator i = keySet.iterator(); i.hasNext(); ) {
String key = (String) i.next();
generator.addProperty(key, (Class) properties.get(key));
}
return generator.create();
}
/**
* 获取属性值的方法
*
* @param obj
* @param property
* @return 属性的值
*/
public static Object getValue(Object obj, String property) {
BeanMap beanMap = BeanMap.create(obj);
return beanMap.get(property);
}
/**
* 更新属性值的方法
*
* @param obj
* @param property
* @param value
*/
public static void setValue(Object obj, String property, Object value) {
BeanMap beanMap = BeanMap.create(obj);
beanMap.put(property, value);
}
/**
* 通过获取String类型的网关schema链接来获取map，用于生成一个Object类型的对象
*
* @param http
* @return 用于反射生成schema类型的对象的一个map集合
*/
public static HashMap<String, Class> getMapFromhttp(String http) {
HashMap<String, Class> map = new HashMap<>();
String schema = HttpClientUtil.requestByGetMethod(http);
Object data = JSON.parseObject(schema).get("data");
//获取fields，并转化为数组，数组的每个元素都是一个name doc type
JSONObject schemaJson = JSON.parseObject(data.toString());
JSONArray fields = (JSONArray) schemaJson.get("fields");
for (Object field : fields) {
String name = JSON.parseObject(field.toString()).get("name").toString();
String type = JSON.parseObject(field.toString()).get("type").toString();
map.put(name, getClassName(type));
}
return map;
}
/**
* 根据http链接获取schema，解析之后返回一个任务列表 (useList toList funcList)
*
* @param http
* @return
*/
public static ArrayList<String[]> getJobListFromHttp(String http) {
ArrayList<String[]> list = new ArrayList<>();
String schema = HttpClientUtil.requestByGetMethod(http);
//解析data
Object data = JSON.parseObject(schema).get("data");
//获取fields，并转化为数组，数组的每个元素都是一个name doc type
JSONObject schemaJson = JSON.parseObject(data.toString());
JSONArray fields = (JSONArray) schemaJson.get("fields");
for (Object field : fields) {
Object doc = JSON.parseObject(field.toString()).get("doc");
String name = JSON.parseObject(field.toString()).get("name").toString();
if (doc != null) {
Object format = JSON.parseObject(doc.toString()).get("format");
if (format != null) {
String functions = null;
String appendTo = null;
String params = null;
Object functionsObj = JSON.parseObject(format.toString()).get("functions");
Object appendToObj = JSON.parseObject(format.toString()).get("appendTo");
Object paramObj = JSON.parseObject(format.toString()).get("param");
if (functionsObj != null) {
functions = functionsObj.toString();
}
if (appendToObj != null) {
appendTo = appendToObj.toString();
}
if (paramObj != null) {
params = paramObj.toString();
}
if (appendTo != null && params == null) {
String[] functionArray = functions.split(",");
String[] appendToArray = appendTo.split(",");
for (int i = 0; i < functionArray.length; i++) {
list.add(new String[]{name, appendToArray[i], functionArray[i], null});
}
} else if (appendTo != null && params != null) {
String[] functionArray = functions.split(",");
String[] appendToArray = appendTo.split(",");
String[] paramArray = params.split(",");
for (int i = 0; i < functionArray.length; i++) {
list.add(new String[]{name, appendToArray[i], functionArray[i], paramArray[i]});
}
} else {
list.add(new String[]{name, name, functions, params});
}
}
}
}
return list;
}
}

64
FlumeDynamicInterceptor/src/main/java/com/zdjizhi/flume/interceptor/utils/system/FlowWriteConfigurations.java Normal file
View File

@@ -0,0 +1,64 @@
package com.zdjizhi.flume.interceptor.utils.system;
import java.util.Properties;
/**
* @author Administrator
*/
public final class FlowWriteConfigurations {
private static Properties propService = new Properties();
public static String getStringProperty(Integer type, String key) {
if (type == 0) {
return propService.getProperty(key);
// } else if (type == 1) {
// return propCommon.getProperty(key);
} else {
return null;
}
}
public static Integer getIntProperty(Integer type, String key) {
if (type == 0) {
return Integer.parseInt(propService.getProperty(key));
// } else if (type == 1) {
// return Integer.parseInt(propCommon.getProperty(key));
} else {
return null;
}
}
public static Long getLongProperty(Integer type, String key) {
if (type == 0) {
return Long.parseLong(propService.getProperty(key));
// } else if (type == 1) {
// return Long.parseLong(propCommon.getProperty(key));
} else {
return null;
}
}
public static Boolean getBooleanProperty(Integer type, String key) {
if (type == 0) {
return "true".equals(propService.getProperty(key).toLowerCase().trim());
// } else if (type == 1) {
// return "true".equals(propCommon.getProperty(key).toLowerCase().trim());
} else {
return null;
}
}
static {
try {
propService.load(FlowWriteConfigurations.class.getClassLoader().getResourceAsStream("service_flow_config.properties"));
} catch (Exception e) {
// propCommon = null;
propService = null;
}
}
}

16
pom.xml Normal file
View File

@@ -0,0 +1,16 @@
<?xml version="1.0" encoding="UTF-8"?>
<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.zdjizhi</groupId>
<artifactId>dynamic_complement</artifactId>
<packaging>pom</packaging>
<version>1.0</version>
<modules>
<module>FlumeDynamicInterceptor</module>
</modules>
</project>