1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package org.codelibs.fess.indexer;
17
18 import java.util.ArrayList;
19 import java.util.List;
20 import java.util.Map;
21 import java.util.function.Consumer;
22
23 import javax.annotation.PostConstruct;
24 import javax.annotation.PreDestroy;
25 import javax.annotation.Resource;
26
27 import org.apache.logging.log4j.LogManager;
28 import org.apache.logging.log4j.Logger;
29 import org.codelibs.core.lang.StringUtil;
30 import org.codelibs.core.lang.ThreadUtil;
31 import org.codelibs.fesen.action.search.SearchRequestBuilder;
32 import org.codelibs.fesen.index.query.QueryBuilder;
33 import org.codelibs.fesen.index.query.QueryBuilders;
34 import org.codelibs.fesen.search.sort.SortOrder;
35 import org.codelibs.fess.Constants;
36 import org.codelibs.fess.crawler.Crawler;
37 import org.codelibs.fess.crawler.entity.AccessResult;
38 import org.codelibs.fess.crawler.entity.AccessResultData;
39 import org.codelibs.fess.crawler.entity.EsAccessResult;
40 import org.codelibs.fess.crawler.entity.EsUrlQueue;
41 import org.codelibs.fess.crawler.service.DataService;
42 import org.codelibs.fess.crawler.service.UrlFilterService;
43 import org.codelibs.fess.crawler.service.UrlQueueService;
44 import org.codelibs.fess.crawler.service.impl.EsDataService;
45 import org.codelibs.fess.crawler.transformer.Transformer;
46 import org.codelibs.fess.crawler.util.EsResultList;
47 import org.codelibs.fess.es.client.SearchEngineClient;
48 import org.codelibs.fess.es.log.exbhv.ClickLogBhv;
49 import org.codelibs.fess.es.log.exbhv.FavoriteLogBhv;
50 import org.codelibs.fess.exception.ContainerNotAvailableException;
51 import org.codelibs.fess.exception.FessSystemException;
52 import org.codelibs.fess.helper.IndexingHelper;
53 import org.codelibs.fess.helper.IntervalControlHelper;
54 import org.codelibs.fess.helper.SearchLogHelper;
55 import org.codelibs.fess.helper.SystemHelper;
56 import org.codelibs.fess.ingest.IngestFactory;
57 import org.codelibs.fess.ingest.Ingester;
58 import org.codelibs.fess.mylasta.direction.FessConfig;
59 import org.codelibs.fess.util.ComponentUtil;
60 import org.codelibs.fess.util.DocList;
61 import org.codelibs.fess.util.MemoryUtil;
62 import org.codelibs.fess.util.ThreadDumpUtil;
63
64 public class IndexUpdater extends Thread {
65 private static final Logger logger = LogManager.getLogger(IndexUpdater.class);
66
67 protected List<String> sessionIdList;
68
69 @Resource
70 protected SearchEngineClient searchEngineClient;
71
72 @Resource
73 protected DataService<EsAccessResult> dataService;
74
75 @Resource
76 protected UrlQueueService<EsUrlQueue> urlQueueService;
77
78 @Resource
79 protected UrlFilterService urlFilterService;
80
81 @Resource
82 protected ClickLogBhv clickLogBhv;
83
84 @Resource
85 protected FavoriteLogBhv favoriteLogBhv;
86
87 @Resource
88 protected SystemHelper systemHelper;
89
90 @Resource
91 protected IndexingHelper indexingHelper;
92
93 protected boolean finishCrawling = false;
94
95 protected long executeTime;
96
97 protected long documentSize;
98
99 protected int maxIndexerErrorCount = 0;
100
101 protected int maxErrorCount = 2;
102
103 protected List<String> finishedSessionIdList = new ArrayList<>();
104
105 private final List<DocBoostMatcher> docBoostMatcherList = new ArrayList<>();
106
107 private List<Crawler> crawlerList;
108
109 private IngestFactory ingestFactory = null;
110
111 public IndexUpdater() {
112
113 }
114
115 @PostConstruct
116 public void init() {
117 if (logger.isDebugEnabled()) {
118 logger.debug("Initialize {}", this.getClass().getSimpleName());
119 }
120 if (ComponentUtil.hasIngestFactory()) {
121 ingestFactory = ComponentUtil.getIngestFactory();
122 }
123 }
124
125 @PreDestroy
126 public void destroy() {
127 if (!finishCrawling) {
128 if (logger.isInfoEnabled()) {
129 logger.info("Stopping all crawler.");
130 }
131 forceStop();
132 }
133 }
134
135 public void addFinishedSessionId(final String sessionId) {
136 synchronized (finishedSessionIdList) {
137 finishedSessionIdList.add(sessionId);
138 }
139 }
140
141 private void deleteBySessionId(final String sessionId) {
142 try {
143 urlFilterService.delete(sessionId);
144 } catch (final Exception e) {
145 logger.warn("Failed to delete url filters: {}", sessionId, e);
146 }
147 try {
148 urlQueueService.delete(sessionId);
149 } catch (final Exception e) {
150 logger.warn("Failed to delete url queues: {}", sessionId, e);
151 }
152 try {
153 dataService.delete(sessionId);
154 } catch (final Exception e) {
155 logger.warn("Failed to delete data: {}", sessionId, e);
156 }
157 }
158
159 @Override
160 public void run() {
161 if (dataService == null) {
162 throw new FessSystemException("DataService is null.");
163 }
164
165 if (logger.isDebugEnabled()) {
166 logger.debug("Starting indexUpdater.");
167 }
168
169 executeTime = 0;
170 documentSize = 0;
171
172 final FessConfig fessConfig = ComponentUtil.getFessConfig();
173 final long updateInterval = fessConfig.getIndexerWebfsUpdateIntervalAsInteger().longValue();
174 final int maxEmptyListCount = fessConfig.getIndexerWebfsMaxEmptyListCountAsInteger();
175 final IntervalControlHelper intervalControlHelper = ComponentUtil.getIntervalControlHelper();
176 try {
177 final Consumer<SearchRequestBuilder> cb = builder -> {
178 final QueryBuilder queryBuilder =
179 QueryBuilders.boolQuery().filter(QueryBuilders.termsQuery(EsAccessResult.SESSION_ID, sessionIdList))
180 .filter(QueryBuilders.termQuery(EsAccessResult.STATUS, org.codelibs.fess.crawler.Constants.OK_STATUS));
181 builder.setQuery(queryBuilder);
182 builder.setFrom(0);
183 final int maxDocumentCacheSize = fessConfig.getIndexerWebfsMaxDocumentCacheSizeAsInteger();
184 builder.setSize(maxDocumentCacheSize <= 0 ? 1 : maxDocumentCacheSize);
185 builder.addSort(EsAccessResult.CREATE_TIME, SortOrder.ASC);
186 };
187
188 final DocList docList = new DocList();
189 final List<EsAccessResult> accessResultList = new ArrayList<>();
190
191 long updateTime = System.currentTimeMillis();
192 int errorCount = 0;
193 int emptyListCount = 0;
194 long cleanupTime = -1;
195 while (!finishCrawling || !accessResultList.isEmpty()) {
196 try {
197 final int sessionIdListSize = finishedSessionIdList.size();
198 intervalControlHelper.setCrawlerRunning(true);
199
200 updateTime = System.currentTimeMillis() - updateTime;
201
202 final long interval = updateInterval - updateTime;
203 if (interval > 0) {
204
205 ThreadUtil.sleep(interval);
206 }
207
208 systemHelper.calibrateCpuLoad();
209
210 docList.clear();
211 accessResultList.clear();
212
213 intervalControlHelper.delayByRules();
214
215 if (logger.isDebugEnabled()) {
216 logger.debug("Processing documents in IndexUpdater queue.");
217 }
218
219 updateTime = System.currentTimeMillis();
220
221 List<EsAccessResult> arList = getAccessResultList(cb, cleanupTime);
222 if (arList.isEmpty()) {
223 emptyListCount++;
224 } else {
225 emptyListCount = 0;
226 }
227 long hitCount = ((EsResultList<EsAccessResult>) arList).getTotalHits();
228 while (hitCount > 0) {
229 if (arList.isEmpty()) {
230 ThreadUtil.sleep(fessConfig.getIndexerWebfsCommitMarginTimeAsInteger().longValue());
231 cleanupTime = -1;
232 } else {
233 processAccessResults(docList, accessResultList, arList);
234 cleanupTime = cleanupAccessResults(accessResultList);
235 }
236 arList = getAccessResultList(cb, cleanupTime);
237 hitCount = ((EsResultList<EsAccessResult>) arList).getTotalHits();
238 }
239 if (!docList.isEmpty()) {
240 indexingHelper.sendDocuments(searchEngineClient, docList);
241 }
242
243 synchronized (finishedSessionIdList) {
244 if (sessionIdListSize != 0 && sessionIdListSize == finishedSessionIdList.size()) {
245 cleanupFinishedSessionData();
246 }
247 }
248 executeTime += System.currentTimeMillis() - updateTime;
249
250 if (logger.isDebugEnabled()) {
251 logger.debug("Processed documents in IndexUpdater queue.");
252 }
253
254
255 errorCount = 0;
256 } catch (final Exception e) {
257 if (errorCount > maxErrorCount) {
258 throw e;
259 }
260 errorCount++;
261 logger.warn("Failed to access data. Retry to access it {} times.", errorCount, e);
262 } finally {
263 if (systemHelper.isForceStop()) {
264 finishCrawling = true;
265 if (logger.isDebugEnabled()) {
266 logger.debug("Stopped indexUpdater.");
267 }
268 }
269 }
270
271 if (emptyListCount >= maxEmptyListCount) {
272 if (logger.isInfoEnabled()) {
273 logger.info("Terminating indexUpdater. emptyListCount is over {}.", maxEmptyListCount);
274 }
275
276 finishCrawling = true;
277 forceStop();
278 if (fessConfig.getIndexerThreadDumpEnabledAsBoolean()) {
279 ThreadDumpUtil.printThreadDump();
280 }
281 org.codelibs.fess.exec.Crawler.addError("QueueTimeout");
282 }
283
284 if (!ComponentUtil.available()) {
285 logger.info("IndexUpdater is terminated.");
286 forceStop();
287 break;
288 }
289 }
290
291 if (logger.isDebugEnabled()) {
292 logger.debug("Finished indexUpdater.");
293 }
294 } catch (final ContainerNotAvailableException e) {
295 if (logger.isDebugEnabled()) {
296 logger.error("IndexUpdater is terminated.", e);
297 } else if (logger.isInfoEnabled()) {
298 logger.info("IndexUpdater is terminated.");
299 }
300 forceStop();
301 } catch (final Throwable t) {
302 if (ComponentUtil.available()) {
303 logger.error("IndexUpdater is terminated.", t);
304 } else if (logger.isDebugEnabled()) {
305 logger.error("IndexUpdater is terminated.", t);
306 org.codelibs.fess.exec.Crawler.addError(t.getClass().getSimpleName());
307 } else if (logger.isInfoEnabled()) {
308 logger.info("IndexUpdater is terminated.");
309 org.codelibs.fess.exec.Crawler.addError(t.getClass().getSimpleName());
310 }
311 forceStop();
312 } finally {
313 intervalControlHelper.setCrawlerRunning(true);
314 }
315
316 if (logger.isInfoEnabled()) {
317 logger.info("[EXEC TIME] index update time: {}ms", executeTime);
318 }
319
320 }
321
322 private void processAccessResults(final DocList docList, final List<EsAccessResult> accessResultList,
323 final List<EsAccessResult> arList) {
324 final FessConfig fessConfig = ComponentUtil.getFessConfig();
325 final long maxDocumentRequestSize = Long.parseLong(fessConfig.getIndexerWebfsMaxDocumentRequestSize());
326 for (final EsAccessResult accessResult : arList) {
327 if (logger.isDebugEnabled()) {
328 logger.debug("Indexing {}", accessResult.getUrl());
329 }
330 accessResult.setStatus(Constants.DONE_STATUS);
331 accessResultList.add(accessResult);
332
333 if (accessResult.getHttpStatusCode() != 200) {
334
335 if (logger.isDebugEnabled()) {
336 logger.debug("Skipped. The response code is {}.", accessResult.getHttpStatusCode());
337 }
338 continue;
339 }
340
341 final long startTime = System.currentTimeMillis();
342 final AccessResultData<?> accessResultData = accessResult.getAccessResultData();
343 if (accessResultData != null) {
344 accessResult.setAccessResultData(null);
345 try {
346 final Transformer transformer = ComponentUtil.getComponent(accessResultData.getTransformerName());
347 if (transformer == null) {
348
349 logger.warn("No transformer: {}", accessResultData.getTransformerName());
350 continue;
351 }
352 @SuppressWarnings("unchecked")
353 final Map<String, Object> map = (Map<String, Object>) transformer.getData(accessResultData);
354 if (map.isEmpty()) {
355
356 logger.warn("No data: {}", accessResult.getUrl());
357 continue;
358 }
359
360 if (Constants.FALSE.equals(map.get(Constants.INDEXING_TARGET))) {
361 if (logger.isDebugEnabled()) {
362 logger.debug("Skipped. This document is not a index target. ");
363 }
364 continue;
365 }
366 map.remove(Constants.INDEXING_TARGET);
367
368 updateDocument(map);
369
370 docList.add(ingest(accessResult, map));
371 final long contentSize = indexingHelper.calculateDocumentSize(map);
372 docList.addContentSize(contentSize);
373 final long processingTime = System.currentTimeMillis() - startTime;
374 docList.addProcessingTime(processingTime);
375 if (logger.isDebugEnabled()) {
376 logger.debug("Added the document({}, {}ms). The number of a document cache is {}.",
377 MemoryUtil.byteCountToDisplaySize(contentSize), processingTime, docList.size());
378 }
379
380 if (docList.getContentSize() >= maxDocumentRequestSize) {
381 indexingHelper.sendDocuments(searchEngineClient, docList);
382 }
383 documentSize++;
384 if (logger.isDebugEnabled()) {
385 logger.debug("The number of an added document is {}.", documentSize);
386 }
387 } catch (final Exception e) {
388 logger.warn("Could not add a doc: {}", accessResult.getUrl(), e);
389 }
390 } else if (logger.isDebugEnabled()) {
391 logger.debug("Skipped. No content. ");
392 }
393
394 }
395 }
396
397 protected Map<String, Object> ingest(final AccessResult<String> accessResult, final Map<String, Object> map) {
398 if (ingestFactory == null) {
399 return map;
400 }
401 Map<String, Object> target = map;
402 for (final Ingester ingester : ingestFactory.getIngesters()) {
403 try {
404 target = ingester.process(target, accessResult);
405 } catch (final Exception e) {
406 logger.warn("Failed to process Ingest[{}]", ingester.getClass().getSimpleName(), e);
407 }
408 }
409 return target;
410 }
411
412 protected void updateDocument(final Map<String, Object> map) {
413 final FessConfig fessConfig = ComponentUtil.getFessConfig();
414
415 if (fessConfig.getIndexerClickCountEnabledAsBoolean()) {
416 addClickCountField(map);
417 }
418
419 if (fessConfig.getIndexerFavoriteCountEnabledAsBoolean()) {
420 addFavoriteCountField(map);
421 }
422
423 float documentBoost = 0.0f;
424 for (final DocBoostMatcher docBoostMatcher : docBoostMatcherList) {
425 if (docBoostMatcher.match(map)) {
426 documentBoost = docBoostMatcher.getValue(map);
427 break;
428 }
429 }
430
431 if (documentBoost > 0) {
432 addBoostValue(map, documentBoost);
433 }
434
435 if (!map.containsKey(fessConfig.getIndexFieldDocId())) {
436 map.put(fessConfig.getIndexFieldDocId(), systemHelper.generateDocId(map));
437 }
438
439 ComponentUtil.getLanguageHelper().updateDocument(map);
440 }
441
442 protected void addBoostValue(final Map<String, Object> map, final float documentBoost) {
443 final FessConfig fessConfig = ComponentUtil.getFessConfig();
444 map.put(fessConfig.getIndexFieldBoost(), documentBoost);
445 if (logger.isDebugEnabled()) {
446 logger.debug("Set a document boost ({}).", documentBoost);
447 }
448 }
449
450 protected void addClickCountField(final Map<String, Object> doc) {
451 final FessConfig fessConfig = ComponentUtil.getFessConfig();
452 final String url = (String) doc.get(fessConfig.getIndexFieldUrl());
453 if (StringUtil.isNotBlank(url)) {
454 final SearchLogHelper searchLogHelper = ComponentUtil.getSearchLogHelper();
455 final int count = searchLogHelper.getClickCount(url);
456 doc.put(fessConfig.getIndexFieldClickCount(), count);
457 if (logger.isDebugEnabled()) {
458 logger.debug("Click Count: {}, url: {}", count, url);
459 }
460 }
461 }
462
463 protected void addFavoriteCountField(final Map<String, Object> map) {
464 final FessConfig fessConfig = ComponentUtil.getFessConfig();
465 final String url = (String) map.get(fessConfig.getIndexFieldUrl());
466 if (StringUtil.isNotBlank(url)) {
467 final SearchLogHelper searchLogHelper = ComponentUtil.getSearchLogHelper();
468 final long count = searchLogHelper.getFavoriteCount(url);
469 map.put(fessConfig.getIndexFieldFavoriteCount(), count);
470 if (logger.isDebugEnabled()) {
471 logger.debug("Favorite Count: {}, url: {}", count, url);
472 }
473 }
474 }
475
476 private long cleanupAccessResults(final List<EsAccessResult> accessResultList) {
477 if (!accessResultList.isEmpty()) {
478 final long execTime = System.currentTimeMillis();
479 final int size = accessResultList.size();
480 dataService.update(accessResultList);
481 accessResultList.clear();
482 final long time = System.currentTimeMillis() - execTime;
483 if (logger.isDebugEnabled()) {
484 logger.debug("Updated {} access results. The execution time is {}ms.", size, time);
485 }
486 return time;
487 }
488 return -1;
489 }
490
491 private List<EsAccessResult> getAccessResultList(final Consumer<SearchRequestBuilder> cb, final long cleanupTime) {
492 if (logger.isDebugEnabled()) {
493 logger.debug("Getting documents in IndexUpdater queue.");
494 }
495 final long execTime = System.currentTimeMillis();
496 final List<EsAccessResult> arList = ((EsDataService) dataService).getAccessResultList(cb);
497 final FessConfig fessConfig = ComponentUtil.getFessConfig();
498 if (!arList.isEmpty()) {
499 final long commitMarginTime = fessConfig.getIndexerWebfsCommitMarginTimeAsInteger().longValue();
500 for (final AccessResult<?> ar : arList.toArray(new AccessResult[arList.size()])) {
501 if (ar.getCreateTime().longValue() > execTime - commitMarginTime) {
502 arList.remove(ar);
503 }
504 }
505 }
506 final long totalHits = ((EsResultList<EsAccessResult>) arList).getTotalHits();
507 if (logger.isInfoEnabled()) {
508 final StringBuilder buf = new StringBuilder(100);
509 buf.append("Processing ");
510 if (totalHits > 0) {
511 buf.append(arList.size()).append('/').append(totalHits).append(" docs (Doc:{access ");
512 } else {
513 buf.append("no docs in indexing queue (Doc:{access ");
514 }
515 buf.append(System.currentTimeMillis() - execTime).append("ms");
516 if (cleanupTime >= 0) {
517 buf.append(", cleanup ").append(cleanupTime).append("ms");
518 }
519 buf.append("}, ");
520 buf.append(MemoryUtil.getMemoryUsageLog());
521 buf.append(')');
522 logger.info(buf.toString());
523 }
524 final long unprocessedDocumentSize = fessConfig.getIndexerUnprocessedDocumentSizeAsInteger().longValue();
525 final IntervalControlHelper intervalControlHelper = ComponentUtil.getIntervalControlHelper();
526 if (totalHits > unprocessedDocumentSize && intervalControlHelper.isCrawlerRunning()) {
527 if (logger.isInfoEnabled()) {
528 logger.info("Stopped all crawler threads. You have {} (>{}) unprocessed docs.", totalHits, unprocessedDocumentSize);
529 }
530 intervalControlHelper.setCrawlerRunning(false);
531 }
532 return arList;
533 }
534
535 private void cleanupFinishedSessionData() {
536 final long execTime = System.currentTimeMillis();
537
538 for (final String sessionId : finishedSessionIdList) {
539 final long execTime2 = System.currentTimeMillis();
540 if (logger.isDebugEnabled()) {
541 logger.debug("Deleting document data: {}", sessionId);
542 }
543 deleteBySessionId(sessionId);
544 if (logger.isDebugEnabled()) {
545 logger.debug("Deleted {} documents. The execution time is {}ms.", sessionId, (System.currentTimeMillis() - execTime2));
546 }
547 }
548 finishedSessionIdList.clear();
549
550 if (logger.isInfoEnabled()) {
551 logger.info("Deleted completed document data. The execution time is {}ms.", (System.currentTimeMillis() - execTime));
552 }
553 }
554
555 private void forceStop() {
556 systemHelper.setForceStop(true);
557 for (final Crawler crawler : crawlerList) {
558 crawler.stop();
559 }
560 }
561
562 public long getExecuteTime() {
563 return executeTime;
564 }
565
566 public List<String> getSessionIdList() {
567 return sessionIdList;
568 }
569
570 public void setSessionIdList(final List<String> sessionIdList) {
571 this.sessionIdList = sessionIdList;
572 }
573
574 public void setFinishCrawling(final boolean finishCrawling) {
575 this.finishCrawling = finishCrawling;
576 }
577
578 public long getDocumentSize() {
579 return documentSize;
580 }
581
582 @Override
583 public void setUncaughtExceptionHandler(final UncaughtExceptionHandler eh) {
584 super.setUncaughtExceptionHandler(eh);
585 }
586
587 public static void setDefaultUncaughtExceptionHandler(final UncaughtExceptionHandler eh) {
588 Thread.setDefaultUncaughtExceptionHandler(eh);
589 }
590
591 public void setMaxIndexerErrorCount(final int maxIndexerErrorCount) {
592 this.maxIndexerErrorCount = maxIndexerErrorCount;
593 }
594
595 public void addDocBoostMatcher(final DocBoostMatcher rule) {
596 docBoostMatcherList.add(rule);
597 }
598
599 public void setCrawlerList(final List<Crawler> crawlerList) {
600 this.crawlerList = crawlerList;
601 }
602 }