|
|
@@ -1,20 +1,21 @@
|
|
|
import cn.reghao.jutil.jdk.serializer.JsonConverter;
|
|
|
-import cn.reghao.jutil.jdk.string.SnowFlake;
|
|
|
-import cn.reghao.tnb.search.app.SearchApplication;
|
|
|
+import cn.reghao.tnb.search.app.config.AppProperties;
|
|
|
import cn.reghao.tnb.search.app.es.*;
|
|
|
import cn.reghao.jutil.jdk.web.log.NginxLog;
|
|
|
+import co.elastic.clients.elasticsearch.ElasticsearchClient;
|
|
|
+import co.elastic.clients.elasticsearch._types.SortOrder;
|
|
|
import co.elastic.clients.elasticsearch._types.mapping.Property;
|
|
|
+import co.elastic.clients.elasticsearch._types.query_dsl.Query;
|
|
|
+import co.elastic.clients.elasticsearch._types.query_dsl.RangeQuery;
|
|
|
+import co.elastic.clients.elasticsearch.indices.AnalyzeRequest;
|
|
|
+import co.elastic.clients.elasticsearch.indices.AnalyzeResponse;
|
|
|
+import co.elastic.clients.elasticsearch.indices.analyze.AnalyzeToken;
|
|
|
+import co.elastic.clients.json.JsonData;
|
|
|
import lombok.extern.slf4j.Slf4j;
|
|
|
import org.junit.jupiter.api.Test;
|
|
|
-import org.springframework.beans.factory.annotation.Autowired;
|
|
|
-import org.springframework.boot.test.context.SpringBootTest;
|
|
|
-import org.springframework.test.context.ActiveProfiles;
|
|
|
|
|
|
-import java.io.File;
|
|
|
-import java.io.FileInputStream;
|
|
|
import java.io.IOException;
|
|
|
-import java.nio.ByteBuffer;
|
|
|
-import java.nio.channels.FileChannel;
|
|
|
+import java.io.RandomAccessFile;
|
|
|
import java.util.ArrayList;
|
|
|
import java.util.List;
|
|
|
import java.util.Map;
|
|
|
@@ -24,154 +25,209 @@ import java.util.Map;
|
|
|
* @date 2025-07-03 15:56:02
|
|
|
*/
|
|
|
@Slf4j
|
|
|
-@ActiveProfiles("dev")
|
|
|
-@SpringBootTest(classes = SearchApplication.class)
|
|
|
public class NginxLogTest {
|
|
|
- @Autowired
|
|
|
- QueryService<NginxLog> queryService;
|
|
|
- @Autowired
|
|
|
- SearchService searchService;
|
|
|
- @Autowired
|
|
|
- DocumentService documentService;
|
|
|
- @Autowired
|
|
|
- IndexService indexService;
|
|
|
- @Autowired
|
|
|
- MappingService mappingService;
|
|
|
- @Autowired
|
|
|
- ElasticService elasticService;
|
|
|
+ ElasticService getElasticService() {
|
|
|
+ AppProperties appProperties = new AppProperties();
|
|
|
+ appProperties.setEsHost("192.168.0.81");
|
|
|
+ appProperties.setEsPort(9200);
|
|
|
+ appProperties.setEsUsername("elastic");
|
|
|
+ appProperties.setEsPassword("VLTtN03SSJ4lsyyg56kf");
|
|
|
+
|
|
|
+ ElasticService elasticService = new ElasticService(appProperties);
|
|
|
+ return elasticService;
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ public void logTest() throws Exception {
|
|
|
+ ElasticService elasticService = getElasticService();
|
|
|
+ SearchService searchService = new SearchService(elasticService);
|
|
|
+
|
|
|
+ int pageSize = 1000;
|
|
|
+ int pageNumber = 1;
|
|
|
+
|
|
|
+ String fieldName = "url.raw";
|
|
|
+ String fieldValue = "/datareceive/ReceiveData/SendContentResult";
|
|
|
+ String sortField = "timeIso8601";
|
|
|
+ //sortField = "requestTime";
|
|
|
+ SortOrder sortOrder = SortOrder.Desc;
|
|
|
+
|
|
|
+ String startDate = "2023-11-08";
|
|
|
+ String endDate = "2024-05-20";
|
|
|
+ String start1 = String.format("%s 00:00:00", startDate);
|
|
|
+ String end1 = String.format("%s 23:59:59", startDate);
|
|
|
+ Query dateQuery0 = RangeQuery.of(q -> q.field("timeIso8601")
|
|
|
+ .gte(JsonData.of(start1)).lte(JsonData.of(end1)).format("yyyy-MM-dd HH:mm:ss"))._toQuery();
|
|
|
+
|
|
|
+ Query combinedQuery = Query.of(q -> q.matchAll(m -> m));
|
|
|
+ if (!fieldValue.isBlank()) {
|
|
|
+ Query termQuery = EsQuery.getTermQuery(fieldName, fieldValue);
|
|
|
+ combinedQuery = dateQuery0;
|
|
|
+ }
|
|
|
+ List<NginxLog> list0 = searchService.searchByPage(indexName, pageSize, pageNumber, combinedQuery, sortField, sortOrder);
|
|
|
+
|
|
|
+ String dateTimeFormat = "yyyy-MM-dd HH:mm:ss";
|
|
|
+ String timeZone = "+08:00";
|
|
|
+ String dateField = "timeIso8601";
|
|
|
+ Query dateQuery = RangeQuery.of(q -> q.field(dateField)
|
|
|
+ .from(start1).to(end1).format(dateTimeFormat).timeZone(timeZone))._toQuery();
|
|
|
+ String aggregateField = "timeIso8601";
|
|
|
+ Map<String, Long> groupMap2 = searchService.aggregateByQuery(indexName, aggregateField, dateQuery);
|
|
|
+ System.out.println();
|
|
|
+ }
|
|
|
+
|
|
|
+ String indexName = "nginx_log";
|
|
|
+ @Test
|
|
|
+ public void aggregateTest() throws Exception {
|
|
|
+ ElasticService elasticService = getElasticService();
|
|
|
+ SearchService searchService = new SearchService(elasticService);
|
|
|
+
|
|
|
+ Query query1 = Query.of(q -> q.matchAll(m -> m));
|
|
|
+ String aggregateField = "timeIso8601";
|
|
|
+ Map<String, Long> groupMap1 = searchService.aggregateByDay(indexName, aggregateField, query1);
|
|
|
+
|
|
|
+ String dateField = "timeIso8601";
|
|
|
+ String startDateTime = "2023-11-09 00:00:00";
|
|
|
+ String endDateTime = "2023-11-09 23:59:59";
|
|
|
+ String start = startDateTime.replace(" ", "T");
|
|
|
+ String end = endDateTime.replace(" ", "T");
|
|
|
+ RangeQuery dateQuery = RangeQuery.of(q -> q.field(dateField).gte(JsonData.of(start)).lte(JsonData.of(end)));
|
|
|
+ Query query = dateQuery._toQuery();
|
|
|
+
|
|
|
+ //aggregateField = "httpUserAgent.raw";
|
|
|
+ aggregateField = "timeIso8601";
|
|
|
+ Map<String, Long> groupMap2 = searchService.aggregateByQuery(indexName, aggregateField, query);
|
|
|
+
|
|
|
+ aggregateField = "methodUrl.raw";
|
|
|
+ Map<String, Long> groupMap3 = searchService.aggregateByQuery(indexName, aggregateField, query);
|
|
|
+
|
|
|
+ /*String windows = "Windows NT";
|
|
|
+ String linux = "Linux ";
|
|
|
+ String mac = "Macintosh";
|
|
|
+ String ipad = "iPad";
|
|
|
+ String iphone = "iPhone";
|
|
|
+ String android = "Android";
|
|
|
+ String client = "client";
|
|
|
+ String others = "others";
|
|
|
+ Map<String, Integer> mapCount = new HashMap<>();
|
|
|
+ mapCount.put(windows, 0);
|
|
|
+ mapCount.put(linux, 0);
|
|
|
+ mapCount.put(mac, 0);
|
|
|
+ mapCount.put(ipad, 0);
|
|
|
+ mapCount.put(iphone, 0);
|
|
|
+ mapCount.put(android, 0);
|
|
|
+ mapCount.put(client, 0);
|
|
|
+ mapCount.put(others, 0);
|
|
|
+
|
|
|
+ groupMap2.keySet().forEach(userAgent -> {
|
|
|
+ if (userAgent.contains(windows)) {
|
|
|
+ incr(windows, mapCount);
|
|
|
+ } else if (userAgent.contains(linux)) {
|
|
|
+ incr(linux, mapCount);
|
|
|
+ } else if (userAgent.contains(mac)) {
|
|
|
+ incr(mac, mapCount);
|
|
|
+ } else if (userAgent.contains(ipad)) {
|
|
|
+ incr(ipad, mapCount);
|
|
|
+ } else if (userAgent.contains(iphone)) {
|
|
|
+ incr(iphone, mapCount);
|
|
|
+ } else if (userAgent.contains(android)) {
|
|
|
+ incr(android, mapCount);
|
|
|
+ } else if (userAgent.contains(client) || userAgent.contains("Client")) {
|
|
|
+ incr(client, mapCount);
|
|
|
+ } else {
|
|
|
+ incr(others, mapCount);
|
|
|
+ }
|
|
|
+ });*/
|
|
|
+ System.out.println();
|
|
|
+ }
|
|
|
+
|
|
|
+ private void incr(String key, Map<String, Integer> map) {
|
|
|
+ map.put(key, map.get(key)+1);
|
|
|
+ }
|
|
|
|
|
|
@Test
|
|
|
- public void indexTest() throws IOException {
|
|
|
- String indexName = "nginx_log";
|
|
|
+ public void esDocTest() {
|
|
|
+ ElasticService elasticService = getElasticService();
|
|
|
+ SearchService searchService = new SearchService(elasticService);
|
|
|
+
|
|
|
+ Query combinedQuery = Query.of(q -> q.matchAll(m -> m));
|
|
|
+ long total = searchService.count(indexName, combinedQuery);
|
|
|
+ log.info("Total documents of {}: {}", indexName, total);
|
|
|
+
|
|
|
+ DocumentService documentService = new DocumentService(elasticService);
|
|
|
+ //documentService.deleteAllDocument(indexName);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ public void esIndexTest() throws IOException {
|
|
|
+ ElasticService elasticService = getElasticService();
|
|
|
+ IndexService indexService = new IndexService(elasticService);
|
|
|
+ MappingService mappingService = new MappingService();
|
|
|
+
|
|
|
indexService.deleteIndex(indexName);
|
|
|
Map<String, Property> propertyMap = mappingService.getPropertyMapWithNginxLog(NginxLog.class);
|
|
|
indexService.createIndex(indexName, propertyMap);
|
|
|
+ }
|
|
|
|
|
|
- //Map<String, Property> propertyMap = mappingService.getVideoTextPropertyMap();
|
|
|
-// indexService.deleteIndex(index);
|
|
|
-// indexService.createIndex(index, propertyMap);
|
|
|
- //indexService.getIndex(index);
|
|
|
- //indexService.deleteIndex(indexName);
|
|
|
- //indexService.getIndex(index);
|
|
|
- //indexService.updateMapping(index);
|
|
|
- //searchService.searchAll(index);
|
|
|
+ @Test
|
|
|
+ public void nginxLogTest() throws IOException {
|
|
|
+ ElasticService elasticService = getElasticService();
|
|
|
+ DocumentService documentService = new DocumentService(elasticService);
|
|
|
+ String filePath = "/home/reghao/Downloads/access-20231107_073356-20240905_165944.log";
|
|
|
+ readFile(filePath, indexName, documentService);
|
|
|
}
|
|
|
|
|
|
- static SnowFlake idGenerator = new SnowFlake(1, 1);
|
|
|
- static long total = 0L;
|
|
|
- public static void readFileFileChannel(File file, String index, DocumentService documentService) {
|
|
|
- List<String> lines = new ArrayList<>();
|
|
|
+ int total = 0;
|
|
|
+ void readFile(String filePath, String index, DocumentService documentService) throws IOException {
|
|
|
List<NginxLog> nginxLogs = new ArrayList<>();
|
|
|
- try {
|
|
|
- FileInputStream fis = new FileInputStream(file);
|
|
|
- FileChannel fileChannel = fis.getChannel();
|
|
|
-
|
|
|
- // 10MB
|
|
|
- int capacity = 10*1024*1024;
|
|
|
- ByteBuffer byteBuffer = ByteBuffer.allocate(capacity);
|
|
|
- StringBuffer buffer = new StringBuffer();
|
|
|
- while(fileChannel.read(byteBuffer) != -1) {
|
|
|
- //读取后，将位置置为0，将limit置为容量, 以备下次读入到字节缓冲中，从0开始存储
|
|
|
- byteBuffer.clear();
|
|
|
- byte[] bytes = byteBuffer.array();
|
|
|
-
|
|
|
- String str = new String(bytes);
|
|
|
- buffer.append(str);
|
|
|
- String[] strArray = buffer.toString().split(System.lineSeparator());
|
|
|
- for (int i = 0; i < strArray.length-1; i++) {
|
|
|
- try {
|
|
|
- NginxLog nginxLog = JsonConverter.jsonToObject(strArray[i], NginxLog.class);
|
|
|
- nginxLogs.add(nginxLog);
|
|
|
- } catch (Exception e) {
|
|
|
- lines.add(strArray[i]);
|
|
|
- //e.printStackTrace();
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- String lastLine = strArray[strArray.length-1];
|
|
|
- if (!lastLine.endsWith("}")) {
|
|
|
- buffer = new StringBuffer();
|
|
|
- buffer.append(strArray[strArray.length-1]);
|
|
|
- }
|
|
|
-
|
|
|
- while (nginxLogs.size() > 10_000) {
|
|
|
- nginxLogs.forEach(nginxLog -> nginxLog.setId(idGenerator.nextId()+""));
|
|
|
- //NginxLog nginxLog = nginxLogs.get(0);
|
|
|
- //documentService.update(index, nginxLog);
|
|
|
-
|
|
|
- documentService.batchAddDocument(index, nginxLogs);
|
|
|
- log.info("save {} nginxLogs", nginxLogs.size());
|
|
|
- total += nginxLogs.size();
|
|
|
- nginxLogs.clear();
|
|
|
- }
|
|
|
+ RandomAccessFile raf = new RandomAccessFile(filePath, "r");
|
|
|
+ String line = raf.readLine();
|
|
|
+ while (line != null) {
|
|
|
+ NginxLog nginxLog0 = JsonConverter.jsonToObject(line, NginxLog.class);
|
|
|
+ nginxLogs.add(nginxLog0);
|
|
|
+
|
|
|
+ if (nginxLogs.size() > 10_000) {
|
|
|
+ total += nginxLogs.size();
|
|
|
+ nginxLogs.forEach(nginxLog -> {
|
|
|
+ String timeIso8601 = nginxLog.getTimeIso8601();
|
|
|
+ // 日期时间格式 yyyy-MM-ddTHH:mm:ss
|
|
|
+ String dateTimeStr = timeIso8601.replace("+08:00", "");
|
|
|
+ nginxLog.setTimeIso8601(dateTimeStr);
|
|
|
+
|
|
|
+ String method = nginxLog.getRequestMethod();
|
|
|
+ String url = nginxLog.getUrl();
|
|
|
+ String methodUrl = String.format("%s %s", method, url);
|
|
|
+ nginxLog.setMethodUrl(methodUrl);
|
|
|
+ });
|
|
|
+
|
|
|
+ documentService.batchAddDocument(index, nginxLogs);
|
|
|
+ log.info("save {} nginxLogs", nginxLogs.size());
|
|
|
+ nginxLogs.clear();
|
|
|
}
|
|
|
- } catch (IOException e) {
|
|
|
- e.printStackTrace();
|
|
|
- } finally {
|
|
|
- // TODO close 处理
|
|
|
- }
|
|
|
- }
|
|
|
|
|
|
- @Test
|
|
|
- public void addNginxLogTest() {
|
|
|
- String index = "nginx_log";
|
|
|
- String baseDir = "/home/reghao/work/azy/data/konglog";
|
|
|
- File dir = new File(baseDir);
|
|
|
- for (File file : dir.listFiles()) {
|
|
|
- readFileFileChannel(file, index, documentService);
|
|
|
- log.info("total -> {}", total);
|
|
|
+ if (total > 1_000_000) {
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ line = raf.readLine();
|
|
|
}
|
|
|
- log.info("total -> {}", total);
|
|
|
}
|
|
|
|
|
|
@Test
|
|
|
- public void nginxLogTest() throws Exception {
|
|
|
- String index = "nginx_log";
|
|
|
- int pageSize = 1000;
|
|
|
- int pageNumber = 1;
|
|
|
- String fieldName = "url.raw";
|
|
|
- String fieldValue = "/datareceive/ReceiveData/SendContentResult";
|
|
|
- //List<NginxLog> list = searchService.searchByPage(index, pageSize, pageNumber, fieldName, fieldValue);
|
|
|
-
|
|
|
- String aggregateField = "url.raw";
|
|
|
- aggregateField = "timeIso8601";
|
|
|
- //searchService.aggregate(index, aggregateField);
|
|
|
-
|
|
|
- //AppProperties elasticProperties = new AppProperties();
|
|
|
- //ElasticService elasticService = new ElasticService(elasticProperties);
|
|
|
- //IndexService indexService = new IndexService(elasticService);
|
|
|
- //MappingService mappingService = new MappingService();
|
|
|
-
|
|
|
- //NginxLogDocument nginxLogDocument = new NginxLogDocument(elasticService);
|
|
|
- //NginxLogSearch nginxLogSearch = new NginxLogSearch(elasticService);
|
|
|
- //QueryService<NginxLog> queryService = new QueryService<>(elasticService);
|
|
|
- //indexService.getIndex(index);
|
|
|
- //indexService.getMapping(index);
|
|
|
-
|
|
|
- //documentService.deleteAllDocument(index);
|
|
|
- //Map<String, Property> propertyMap = mappingService.getPropertyMapWithNginxLog(NginxLog.class);
|
|
|
- //indexService.deleteIndex(index);
|
|
|
- //indexService.createIndex(index, propertyMap);
|
|
|
-
|
|
|
- //searchService.search(index);
|
|
|
-// search1(esClient, "app_log");
|
|
|
-// index = "app_log";
|
|
|
-// deleteAll(index);
|
|
|
-// searchService.aggregate(index);
|
|
|
-
|
|
|
- /*int pn = 1;
|
|
|
- while (pn < 100) {
|
|
|
- List<NginxLog> list = searchService.searchByPage(index, pn, "", "");
|
|
|
- System.out.println();
|
|
|
- pn++;
|
|
|
- }*/
|
|
|
-
|
|
|
- //searchService.searchAll(index);
|
|
|
-// indexService.updateMapping(index);
|
|
|
-
|
|
|
- //String queryString = "content";
|
|
|
- //List<NginxLog> list = queryService.queryWithHighlight(index, queryString, pn, ps, NginxLog.class);
|
|
|
- //searchService.searchAll(index);
|
|
|
- //nginxLogSearch.aggregate();
|
|
|
+ public void analyzerTest() throws IOException {
|
|
|
+ ElasticService elasticService = getElasticService();
|
|
|
+ // es 默认的标准分词器
|
|
|
+ String analyzer = "standard";
|
|
|
+ // ik 的两种分词器
|
|
|
+ //analyzer = "ik_max_word";
|
|
|
+ analyzer = "ik_smart";
|
|
|
+
|
|
|
+ String text = "中华人民共和国国歌";
|
|
|
+ AnalyzeRequest analyzeRequest = new AnalyzeRequest.Builder()
|
|
|
+ .text(text)
|
|
|
+ .analyzer(analyzer)
|
|
|
+ .build();
|
|
|
+ ElasticsearchClient esClient = elasticService.getElasticsearchClient();
|
|
|
+ AnalyzeResponse analyzeResponse = esClient.indices().analyze(analyzeRequest);
|
|
|
+ List<AnalyzeToken> tokens = analyzeResponse.tokens();
|
|
|
+ System.out.println();
|
|
|
}
|
|
|
}
|
