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

增加radius写入hbase 和 radius上线下功能代码

Browse Source
This commit is contained in:
qidaijie
2020-04-14 10:41:36 +08:00
parent 9bad398382
commit c5209d10e2
12 changed files with 1005 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

145
dynamic_complement/FlumeRadiusOnOffInterceptor/pom.xml Normal file
View File

@@ -0,0 +1,145 @@
<?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>FlumeRadiusOnOffInterceptor</artifactId>
<repositories>
<repository>
<id>nexus</id>
<name>Team Nexus Repository</name>
<url>http://192.168.40.125:8099/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>
</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.FlumeOnOffApp</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.FlumeOnOffApp</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>
</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.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.47</version>
</dependency>
<dependency>
<groupId>cglib</groupId>
<artifactId>cglib-nodep</artifactId>
<version>3.2.4</version>
</dependency>
<dependency>
<groupId>com.zdjizhi</groupId>
<artifactId>galaxy</artifactId>
<version>1.0.2</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>
</dependencies>
</project>

16
dynamic_complement/FlumeRadiusOnOffInterceptor/properties/service_flow_config.properties Normal file
View File

@@ -0,0 +1,16 @@
#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
#hbase-zookeeper地址
hbase.zookeeper.servers=192.168.40.224:2181
#hbase表名
hbase.table.name=subscriber_info

145
dynamic_complement/FlumeRadiusOnOffInterceptor/src/main/java/com/zdjizhi/flume/interceptor/FlumeOnOffApp.java Normal file
View File

@@ -0,0 +1,145 @@
package com.zdjizhi.flume.interceptor;
import com.alibaba.fastjson.JSONObject;
import com.google.common.base.Preconditions;
import com.zdjizhi.flume.interceptor.bean.Knowledge;
import com.zdjizhi.flume.interceptor.common.OnOffConfig;
import com.zdjizhi.utils.StringUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import org.apache.log4j.Logger;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.*;
/**
* @author qidaijie
*/
public class FlumeOnOffApp implements Interceptor {
private static Logger logger = Logger.getLogger(FlumeOnOffApp.class);
@Override
public void initialize() {
}
@Override
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 {
message = parsingMessage(message);
if (StringUtils.isNotBlank(message)) {
event.setBody(message.getBytes());
return event;
} else {
return null;
}
} catch (Exception e) {
logger.error("FlumeOnOffApp intercept(Event event) method is error===>{" + e + "}<===");
e.printStackTrace();
}
return null;
}
@Override
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;
}
@Override
public void close() {
logger.warn("FlumeOnOffApp is closed.");
}
/**
* 解析日志，并补全
* 补domain,补subscriber_id
*
* @param message Security原始日志
* @return 补全后的日志
* <p>
*/
private String parsingMessage(String message) {
if (StringUtil.isNotBlank(message)) {
JSONObject jsonObject = JSONObject.parseObject(message);
//数据需包含 radius_packet_type and radius_acct_status_type 字段
if (jsonObject.containsKey(OnOffConfig.RADIUS_PACKET_TYPE) && jsonObject.containsKey(OnOffConfig.RADIUS_ACCT_STATUS_TYPE)) {
int packetType = jsonObject.getInteger(OnOffConfig.RADIUS_PACKET_TYPE);
int statusType = jsonObject.getInteger(OnOffConfig.RADIUS_ACCT_STATUS_TYPE);
//条件radius_packet_type = 4 and radius_acct_status_type = 1 or 2
// boolean existed = OnOffConfig.ACCOUNTING_REQUEST == packetType && (OnOffConfig.START_BILLING == statusType || OnOffConfig.STOP_BILLING == statusType);
if (OnOffConfig.ACCOUNTING_REQUEST == packetType && (OnOffConfig.START_BILLING == statusType || OnOffConfig.STOP_BILLING == statusType)) {
Knowledge knowledge = new Knowledge();
knowledge.setFramed_ip(jsonObject.getString("radius_framed_ip"));
knowledge.setAccount(jsonObject.getString("radius_account"));
knowledge.setAcct_status_type(statusType);
/*
*如果存在时间戳则选择此时间戳没有获取当前时间
*/
if (jsonObject.containsKey(OnOffConfig.RADIUS_EVENT_TIMESTAMP)) {
knowledge.setEvent_timestamp(jsonObject.getInteger("radius_event_timestamp"));
} else {
knowledge.setEvent_timestamp((System.currentTimeMillis() / 1000));
}
/*
* 标识同一个连接：
* 1.数据若存在acct_multi_session_id属性，取该属性
* 2. 不存在取 acct_session_id
*/
if (jsonObject.containsKey(OnOffConfig.RADIUS_MULTI_SESSION_ID)) {
knowledge.setAcct_session_id(jsonObject.getString("radius_acct_multi_session_id"));
} else {
knowledge.setAcct_session_id(jsonObject.getString("radius_acct_session_id"));
}
/*
*用户的在线时长，以秒为单位,下线用户无此属性，默认为0
*/
if (jsonObject.containsKey(OnOffConfig.RADIUS_SESSION_TIME)) {
knowledge.setAcct_session_time(jsonObject.getInteger("radius_acct_session_time"));
} else {
knowledge.setAcct_session_time(0);
}
return JSONObject.toJSONString(knowledge);
}
}
}
return null;
}
public static class FlumeDynamicAppBuilder implements Interceptor.Builder {
@Override
public Interceptor build() {
return new FlumeOnOffApp();
}
@Override
public void configure(Context context) {
}
}
}

61
dynamic_complement/FlumeRadiusOnOffInterceptor/src/main/java/com/zdjizhi/flume/interceptor/bean/Knowledge.java Normal file
View File

@@ -0,0 +1,61 @@
package com.zdjizhi.flume.interceptor.bean;
/**
* @author qidaijie
*/
public class Knowledge {
private String framed_ip;
private String account;
private String acct_session_id;
private int acct_status_type;
private int acct_session_time;
private long event_timestamp;
public String getFramed_ip() {
return framed_ip;
}
public void setFramed_ip(String framed_ip) {
this.framed_ip = framed_ip;
}
public String getAccount() {
return account;
}
public void setAccount(String account) {
this.account = account;
}
public int getAcct_status_type() {
return acct_status_type;
}
public void setAcct_status_type(int acct_status_type) {
this.acct_status_type = acct_status_type;
}
public long getEvent_timestamp() {
return event_timestamp;
}
public void setEvent_timestamp(long event_timestamp) {
this.event_timestamp = event_timestamp;
}
public String getAcct_session_id() {
return acct_session_id;
}
public void setAcct_session_id(String acct_session_id) {
this.acct_session_id = acct_session_id;
}
public int getAcct_session_time() {
return acct_session_time;
}
public void setAcct_session_time(int acct_session_time) {
this.acct_session_time = acct_session_time;
}
}

50
dynamic_complement/FlumeRadiusOnOffInterceptor/src/main/java/com/zdjizhi/flume/interceptor/common/OnOffConfig.java Normal file
View File

@@ -0,0 +1,50 @@
package com.zdjizhi.flume.interceptor.common;
/**
* @author Administrator
*/
public class OnOffConfig {
/**
* 4- Accounting-Request(账户授权)
*/
public static final int ACCOUNTING_REQUEST = 4;
/**
* 1、开始计费
*/
public static final int START_BILLING = 1;
/**
* 2、停止计费
*/
public static final int STOP_BILLING = 2;
/**
* 计费请求报文类型
*/
public static final String RADIUS_ACCT_STATUS_TYPE = "radius_acct_status_type";
/**
* 报文类型
*/
public static final String RADIUS_PACKET_TYPE = "radius_packet_type";
/**
* 发送计费请求报文时间戳
*/
public static final String RADIUS_EVENT_TIMESTAMP = "radius_event_timestamp";
/**
* 一个用户多个计费ID关联属性
*/
public static final String RADIUS_MULTI_SESSION_ID = "radius_acct_multi_session_id";
/**
* 用户的在线时长，以秒为单位
*/
public static final String RADIUS_SESSION_TIME = "radius_acct_session_time";
/**
* flume使用配置
*/
public static final String HBASE_ZOOKEEPER_SERVERS = OnOffConfigurations.getStringProperty(0, "hbase.zookeeper.servers");
public static final String HBASE_TABLE_NAME = OnOffConfigurations.getStringProperty(0, "hbase.table.name");
}

55
dynamic_complement/FlumeRadiusOnOffInterceptor/src/main/java/com/zdjizhi/flume/interceptor/common/OnOffConfigurations.java Normal file
View File

@@ -0,0 +1,55 @@
package com.zdjizhi.flume.interceptor.common;
import java.util.Properties;
/**
* @author Administrator
*/
public final class OnOffConfigurations {
private static Properties propService = new Properties();
public static String getStringProperty(Integer type, String key) {
if (type == 0) {
return propService.getProperty(key);
} else {
return null;
}
}
public static Integer getIntProperty(Integer type, String key) {
if (type == 0) {
return Integer.parseInt(propService.getProperty(key));
} else {
return null;
}
}
public static Long getLongProperty(Integer type, String key) {
if (type == 0) {
return Long.parseLong(propService.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 {
return null;
}
}
static {
try {
propService.load(OnOffConfigurations.class.getClassLoader().getResourceAsStream("service_flow_config.properties"));
} catch (Exception e) {
propService = null;
}
}
}

163
dynamic_complement/FlumeSubscriberInterceptor/pom.xml Normal file
View File

@@ -0,0 +1,163 @@
<?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>FlumeSubscriberInterceptor</artifactId>
<repositories>
<!--<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>
</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.FlumeSubscriberInterceptor</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.FlumeSubscriberApp</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>
</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.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.47</version>
</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>
</dependencies>
</project>

16
dynamic_complement/FlumeSubscriberInterceptor/properties/service_flow_config.properties Normal file
View File

@@ -0,0 +1,16 @@
#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
#hbase-zookeeper地址
hbase.zookeeper.servers=192.168.40.224:2181
#hbase表名
hbase.table.name=subscriber_info

248
dynamic_complement/FlumeSubscriberInterceptor/src/main/java/com/zdjizhi/flume/interceptor/FlumeSubscriberApp.java Normal file
View File

@@ -0,0 +1,248 @@
package com.zdjizhi.flume.interceptor;
import com.alibaba.fastjson.JSONObject;
import com.google.common.base.Preconditions;
import com.zdjizhi.flume.interceptor.common.SubscriberConfig;
import org.apache.commons.lang.StringUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
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.io.UnsupportedEncodingException;
import java.util.*;
/**
* @author qidaijie
*/
public class FlumeSubscriberApp implements Interceptor {
private static Logger logger = Logger.getLogger(FlumeSubscriberApp.class);
private static Map<String, String> subIdMap;
private List<Put> putList;
private static Connection connection;
private String hbaseZookeeperIp;
private String hbaseTableName;
public FlumeSubscriberApp(String hbaseZookeeperIp, String hbaseTableName) {
this.hbaseZookeeperIp = hbaseZookeeperIp;
this.hbaseTableName = hbaseTableName;
}
@Override
public void initialize() {
subIdMap = new HashMap<>(256);
putList = new ArrayList<>();
// 管理HBase的配置信息
Configuration configuration = HBaseConfiguration.create();
// 设置zookeeper节点
configuration.set("hbase.zookeeper.quorum", hbaseZookeeperIp);
configuration.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
configuration.set("hbase.client.retries.number", "3");
configuration.set("hbase.bulkload.retries.number", "3");
configuration.set("zookeeper.recovery.retry", "3");
try {
connection = ConnectionFactory.createConnection(configuration);
} catch (IOException e) {
logger.error("用户名信息写入HBase程序，连接HBase异常");
e.printStackTrace();
}
getAll(hbaseTableName);
new Timer().schedule(new TimerTask() {
@Override
public void run() {
try {
insertData(putList, hbaseTableName);
} catch (Exception e) {
e.printStackTrace();
}
}
}, 0, 50000);
}
@Override
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("FlumeSubscriberApp intercept(Event event) method is error===>{" + e + "}<===");
e.printStackTrace();
}
return event;
}
@Override
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;
}
@Override
public void close() {
logger.warn("FlumeSubscriberApp is closed.");
}
/**
* 解析日志，并补全
* 补domain,补subscriber_id
*
* @param message Security原始日志
* @return 补全后的日志
* <p>
*/
private String dealCommonMessage(String message) {
JSONObject jsonObject = JSONObject.parseObject(message);
if (jsonObject.containsKey(SubscriberConfig.PACKET_TYPE) && jsonObject.containsKey(SubscriberConfig.STATUS_TYPE)) {
if (SubscriberConfig.ACCOUNTING_REQUEST == jsonObject.getInteger(SubscriberConfig.PACKET_TYPE)
&& SubscriberConfig.START_BILLING == jsonObject.getInteger(SubscriberConfig.STATUS_TYPE)) {
String framedIp = jsonObject.getString("radius_framed_ip");
String account = jsonObject.getString("radius_account");
dataValidation(framedIp, account, putList);
}
if (putList.size() == SubscriberConfig.LIST_SIZE_MAX) {
insertData(putList, hbaseTableName);
}
}
return message;
}
/**
* 获取所有的 key value
*/
private static void getAll(String tableNmae) {
try {
Table table = connection.getTable(TableName.valueOf("sub:" + tableNmae));
Scan scan = new Scan();
ResultScanner scanner = table.getScanner(scan);
for (Result result : scanner) {
Cell[] cells = result.rawCells();
for (Cell cell : cells) {
subIdMap.put(Bytes.toString(CellUtil.cloneRow(cell)), Bytes.toString(CellUtil.cloneValue(cell)));
}
}
scanner.close();
} catch (IOException e) {
logger.error("获取HBase所有row key出现异常");
e.printStackTrace();
}
}
/**
* 写入数据到HBase
*
* @param putList puts list
*/
private static void insertData(List<Put> putList, String tableName) {
Table table = null;
try {
table = connection.getTable(TableName.valueOf("sub:" + tableName));
table.put(putList);
putList.clear();
} catch (IOException e) {
e.printStackTrace();
} finally {
try {
if (table != null) {
table.close();
}
} catch (IOException e) {
logger.error("更新数据写入HBase失败");
e.printStackTrace();
}
}
}
/**
* 验证数据并与内存中的对比
*
* @param ip framed_ip
* @param account account
*/
private static void dataValidation(String ip, String account, List<Put> putList) {
if (subIdMap.containsKey(ip)) {
if (!subIdMap.get(ip).equals(account)) {
Put put = new Put(ip.getBytes());
put.addColumn("subscriber_id".getBytes(), "account".getBytes(), account.getBytes());
putList.add(put);
subIdMap.put(ip, account);
}
} else {
Put put = new Put(ip.getBytes());
put.addColumn("subscriber_id".getBytes(), "account".getBytes(), account.getBytes());
putList.add(put);
subIdMap.put(ip, account);
}
}
public static class FlumeDynamicAppBuilder implements Interceptor.Builder {
private String hbaseZookeeperIp;
private String hbaseTableName;
@Override
public Interceptor build() {
return new FlumeSubscriberApp(hbaseZookeeperIp, hbaseTableName);
}
@Override
public void configure(Context context) {
try {
this.hbaseZookeeperIp = context.getString("hbaseZookeeperIp", "");
Preconditions.checkNotNull("".equals(hbaseZookeeperIp), "hbaseZookeeperIp must be set!!");
logger.info("FlumeSubscriberApp Read hbaseZookeeperIp from configuration : " + hbaseZookeeperIp);
} catch (IllegalArgumentException e) {
throw new IllegalArgumentException("FlumeSubscriberApp hbaseZookeeperIp invalid", e);
} catch (Exception e) {
logger.error("FlumeSubscriberApp Get hbaseZookeeperIp is error : " + e);
}
try {
this.hbaseTableName = context.getString("hbaseTableName", "");
Preconditions.checkNotNull("".equals(hbaseTableName), "hbaseTableName must be set!!");
logger.info("FlumeSubscriberApp Read hbaseTableName from configuration : " + hbaseTableName);
} catch (IllegalArgumentException e) {
throw new IllegalArgumentException("FlumeSubscriberApp hbaseTableName invalid", e);
} catch (Exception e) {
logger.error("FlumeSubscriberApp Get hbaseTableName is error : " + e);
}
}
}
}

34
dynamic_complement/FlumeSubscriberInterceptor/src/main/java/com/zdjizhi/flume/interceptor/common/SubscriberConfig.java Normal file
View File

@@ -0,0 +1,34 @@
package com.zdjizhi.flume.interceptor.common;
/**
* @author Administrator
*/
public class SubscriberConfig {
/**
* 最多存在多少数据即写入hbase
*/
public static final int LIST_SIZE_MAX = 5000;
/**
* 4- Accounting-Request(账户授权)
*/
public static final int ACCOUNTING_REQUEST = 4;
/**
* radius_packet_type
*/
public static final String PACKET_TYPE = "radius_packet_type";
/**
* 1、开始计费
*/
public static final int START_BILLING = 1;
/**
* radius_acct_status_type
*/
public static final String STATUS_TYPE = "radius_acct_status_type";
/**
* flume使用配置
*/
public static final String HBASE_ZOOKEEPER_SERVERS = SubscriberConfigurations.getStringProperty(0, "hbase.zookeeper.servers");
public static final String HBASE_TABLE_NAME = SubscriberConfigurations.getStringProperty(0, "hbase.table.name");
}

55
dynamic_complement/FlumeSubscriberInterceptor/src/main/java/com/zdjizhi/flume/interceptor/common/SubscriberConfigurations.java Normal file
View File

@@ -0,0 +1,55 @@
package com.zdjizhi.flume.interceptor.common;
import java.util.Properties;
/**
* @author Administrator
*/
public final class SubscriberConfigurations {
private static Properties propService = new Properties();
public static String getStringProperty(Integer type, String key) {
if (type == 0) {
return propService.getProperty(key);
} else {
return null;
}
}
public static Integer getIntProperty(Integer type, String key) {
if (type == 0) {
return Integer.parseInt(propService.getProperty(key));
} else {
return null;
}
}
public static Long getLongProperty(Integer type, String key) {
if (type == 0) {
return Long.parseLong(propService.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 {
return null;
}
}
static {
try {
propService.load(SubscriberConfigurations.class.getClassLoader().getResourceAsStream("service_flow_config.properties"));
} catch (Exception e) {
propService = null;
}
}
}

17
dynamic_complement/pom.xml Normal file
View File

@@ -0,0 +1,17 @@
<?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>FlumeSubscriberInterceptor</module>
<module>FlumeRadiusOnOffInterceptor</module>
</modules>
</project>