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