1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package org.codelibs.fess.timer;
17
18 import java.util.HashMap;
19 import java.util.List;
20 import java.util.Map;
21
22 import org.apache.logging.log4j.LogManager;
23 import org.apache.logging.log4j.Logger;
24 import org.codelibs.core.timer.TimeoutTarget;
25 import org.codelibs.fess.helper.LogNotificationHelper;
26 import org.codelibs.fess.helper.LogNotificationHelper.LogNotificationEvent;
27 import org.codelibs.fess.opensearch.client.SearchEngineClient;
28 import org.codelibs.fess.util.ComponentUtil;
29 import org.opensearch.action.bulk.BulkRequestBuilder;
30 import org.opensearch.action.bulk.BulkResponse;
31
32
33
34
35
36 public class LogNotificationTarget implements TimeoutTarget {
37
38
39
40
41 public LogNotificationTarget() {
42
43 }
44
45 private static final Logger logger = LogManager.getLogger(LogNotificationTarget.class);
46 private static final String NOTIFICATION_QUEUE_INDEX = "fess_log.notification_queue";
47 private static final int BATCH_SIZE = 100;
48 private volatile boolean indexChecked = false;
49
50 @Override
51 public void expired() {
52 try {
53 if (!ComponentUtil.getFessConfig().isLogNotificationEnabled()) {
54 return;
55 }
56 } catch (final Exception e) {
57 return;
58 }
59
60 final LogNotificationHelper helper;
61 try {
62 helper = ComponentUtil.getLogNotificationHelper();
63 } catch (final Exception e) {
64 return;
65 }
66
67 final List<LogNotificationEvent> events = helper.drainAll();
68 if (events.isEmpty()) {
69 return;
70 }
71
72 try {
73 final SearchEngineClient client = ComponentUtil.getSearchEngineClient();
74 final String hostname = ComponentUtil.getSystemHelper().getHostname();
75 final String indexName = resolveIndexName();
76
77 ensureIndexExists(client, indexName);
78
79 for (int i = 0; i < events.size(); i += BATCH_SIZE) {
80 final List<LogNotificationEvent> batch = events.subList(i, Math.min(i + BATCH_SIZE, events.size()));
81 final BulkRequestBuilder bulkRequest = client.prepareBulk();
82 for (final LogNotificationEvent event : batch) {
83 final Map<String, Object> source = new HashMap<>();
84 source.put("hostname", hostname);
85 source.put("level", event.getLevel());
86 source.put("loggerName", event.getLoggerName());
87 source.put("message", event.getMessage());
88 source.put("throwable", event.getThrowable() != null ? event.getThrowable() : "");
89 source.put("timestamp", event.getTimestamp());
90 bulkRequest.add(client.prepareIndex().setIndex(indexName).setSource(source));
91 }
92 final BulkResponse response = bulkRequest.execute().actionGet(30_000L);
93 if (response.hasFailures()) {
94 logger.warn("Failed to write log notifications: {}", response.buildFailureMessage());
95 }
96 }
97 } catch (final Exception e) {
98 logger.debug("Failed to flush log notifications to OpenSearch.", e);
99 }
100 }
101
102
103
104
105 public void flush() {
106 expired();
107 }
108
109 private String resolveIndexName() {
110 return ComponentUtil.getFessConfig().getIndexLogIndex() + ".notification_queue";
111 }
112
113 private void ensureIndexExists(final SearchEngineClient client, final String indexName) {
114 if (!indexChecked) {
115 if (!client.existsIndex(indexName)) {
116 client.createIndex(NOTIFICATION_QUEUE_INDEX, indexName);
117 }
118 indexChecked = true;
119 }
120 }
121 }