1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package org.codelibs.fess.ds.callback;
17
18 import static org.codelibs.core.stream.StreamUtil.stream;
19
20 import java.util.ArrayList;
21 import java.util.Deque;
22 import java.util.HashSet;
23 import java.util.LinkedList;
24 import java.util.List;
25 import java.util.Map;
26 import java.util.Objects;
27 import java.util.Set;
28 import java.util.concurrent.ExecutorService;
29 import java.util.concurrent.LinkedBlockingQueue;
30 import java.util.concurrent.ThreadPoolExecutor;
31 import java.util.concurrent.TimeUnit;
32 import java.util.regex.Pattern;
33 import java.util.stream.Collectors;
34
35 import org.apache.logging.log4j.LogManager;
36 import org.apache.logging.log4j.Logger;
37 import org.codelibs.core.lang.StringUtil;
38 import org.codelibs.fess.Constants;
39 import org.codelibs.fess.crawler.builder.RequestDataBuilder;
40 import org.codelibs.fess.crawler.client.CrawlerClient;
41 import org.codelibs.fess.crawler.client.CrawlerClientFactory;
42 import org.codelibs.fess.crawler.entity.ResponseData;
43 import org.codelibs.fess.crawler.entity.ResultData;
44 import org.codelibs.fess.crawler.exception.ChildUrlsException;
45 import org.codelibs.fess.crawler.exception.CrawlerSystemException;
46 import org.codelibs.fess.crawler.processor.ResponseProcessor;
47 import org.codelibs.fess.crawler.processor.impl.DefaultResponseProcessor;
48 import org.codelibs.fess.crawler.rule.Rule;
49 import org.codelibs.fess.crawler.rule.RuleManager;
50 import org.codelibs.fess.crawler.serializer.DataSerializer;
51 import org.codelibs.fess.crawler.transformer.Transformer;
52 import org.codelibs.fess.entity.DataStoreParams;
53 import org.codelibs.fess.exception.DataStoreCrawlingException;
54 import org.codelibs.fess.helper.CrawlerStatsHelper;
55 import org.codelibs.fess.helper.CrawlerStatsHelper.StatsAction;
56 import org.codelibs.fess.helper.CrawlerStatsHelper.StatsKeyObject;
57 import org.codelibs.fess.helper.IndexingHelper;
58 import org.codelibs.fess.helper.SystemHelper;
59 import org.codelibs.fess.mylasta.direction.FessConfig;
60 import org.codelibs.fess.opensearch.client.SearchEngineClient;
61 import org.codelibs.fess.util.ComponentUtil;
62 import org.lastaflute.di.core.SingletonLaContainer;
63 import org.opensearch.index.query.QueryBuilders;
64
65
66
67
68
69
70
71
72
73
74 public class FileListIndexUpdateCallbackImpl implements IndexUpdateCallback {
75
76
77 private static final Logger logger = LogManager.getLogger(FileListIndexUpdateCallbackImpl.class);
78
79
80
81
82
83
84 protected static final String URL_EXCLUDE_PATTERN = "url_exclude_pattern";
85
86
87 protected IndexUpdateCallback indexUpdateCallback;
88
89
90 protected CrawlerClientFactory crawlerClientFactory;
91
92
93
94
95
96 protected List<String> deleteUrlList = new ArrayList<>();
97
98
99 protected int maxDeleteDocumentCacheSize;
100
101
102 protected int maxRedirectCount;
103
104
105 private final ExecutorService executor;
106
107
108 private int executorTerminationTimeout = 300;
109
110
111
112
113
114
115
116
117 public FileListIndexUpdateCallbackImpl(final IndexUpdateCallback indexUpdateCallback, final CrawlerClientFactory crawlerClientFactory,
118 final int nThreads) {
119 this.indexUpdateCallback = indexUpdateCallback;
120 this.crawlerClientFactory = crawlerClientFactory;
121 executor = newFixedThreadPool(nThreads < 1 ? 1 : nThreads);
122 final FessConfig fessConfig = ComponentUtil.getFessConfig();
123 maxDeleteDocumentCacheSize = fessConfig.getIndexerDataMaxDeleteCacheSizeAsInteger();
124 maxRedirectCount = fessConfig.getIndexerDataMaxRedirectCountAsInteger();
125 }
126
127
128
129
130
131
132
133 protected ExecutorService newFixedThreadPool(final int nThreads) {
134 if (logger.isDebugEnabled()) {
135 logger.debug("Initialized executor thread pool: size={}", nThreads);
136 }
137 return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(nThreads),
138 new ThreadPoolExecutor.CallerRunsPolicy());
139 }
140
141 @Override
142 public void store(final DataStoreParams paramMap, final Map<String, Object> dataMap) {
143 final CrawlerStatsHelper crawlerStatsHelper = ComponentUtil.getCrawlerStatsHelper();
144 final StatsKeyObject keyObj = paramMap.get(Constants.CRAWLER_STATS_KEY) instanceof final StatsKeyObject sko ? sko : null;
145 if (keyObj != null) {
146 crawlerStatsHelper.runOnThread(keyObj);
147 }
148 final DataStoreParams localParams = paramMap.newInstance();
149 executor.execute(() -> {
150 try {
151 final Object eventType = dataMap.remove(getParamValue(localParams, "field.event_type", "event_type"));
152 if (getParamValue(localParams, "event.create", "create").equals(eventType)
153 || getParamValue(localParams, "event.modify", "modify").equals(eventType)) {
154
155 addDocument(localParams, dataMap);
156 } else if (getParamValue(localParams, "event.delete", "delete").equals(eventType)) {
157
158 deleteDocument(localParams, dataMap);
159 } else {
160 logger.warn("Unknown event type: '{}'. Supported: [create, modify, delete]. url={}", eventType,
161 dataMap.get(ComponentUtil.getFessConfig().getIndexFieldUrl()));
162 }
163 } finally {
164 if (keyObj != null) {
165 crawlerStatsHelper.done(keyObj);
166 }
167 }
168 });
169 }
170
171
172
173
174
175
176
177
178
179 protected String getParamValue(final DataStoreParams paramMap, final String key, final String defaultValue) {
180 return paramMap.getAsString(key, defaultValue);
181 }
182
183
184
185
186
187
188
189
190
191 protected void addDocument(final DataStoreParams paramMap, final Map<String, Object> dataMap) {
192 final FessConfig fessConfig = ComponentUtil.getFessConfig();
193 final CrawlerStatsHelper crawlerStatsHelper = ComponentUtil.getCrawlerStatsHelper();
194 synchronized (indexUpdateCallback) {
195
196 if (!dataMap.containsKey(fessConfig.getIndexFieldUrl()) || dataMap.get(fessConfig.getIndexFieldUrl()) == null) {
197 logger.warn("Could not add document: url field is missing or null");
198 return;
199 }
200
201 final String url = dataMap.get(fessConfig.getIndexFieldUrl()).toString();
202 final CrawlerClient client = crawlerClientFactory.getClient(url);
203 if (client == null) {
204 logger.warn("CrawlerClient not available for url='{}'. Protocol may not be supported.", url);
205 return;
206 }
207
208 final StatsKeyObject keyObj = paramMap.get(Constants.CRAWLER_STATS_KEY) instanceof final StatsKeyObject sko ? sko : null;
209
210 final long maxAccessCount = getMaxAccessCount(paramMap, dataMap);
211 final int maxDepth = getMaxDepth(paramMap, dataMap);
212 long counter = 0;
213 final Deque<CrawlRequest> requestQueue = new LinkedList<>();
214 final Set<String> processedUrls = new HashSet<>();
215 requestQueue.offer(new CrawlRequest(url, 0));
216 while (!requestQueue.isEmpty() && (maxAccessCount < 0 || counter < maxAccessCount)) {
217 final CrawlRequest crawlRequest = requestQueue.poll();
218 if ((maxDepth != -1 && crawlRequest.getDepth() > maxDepth) || processedUrls.contains(crawlRequest.getUrl())) {
219 if (logger.isDebugEnabled()) {
220 logger.debug("Skipping crawl request for url='{}' at depth={} (maxDepth={}, alreadyProcessed={})",
221 crawlRequest.getUrl(), crawlRequest.getDepth(), maxDepth, processedUrls.contains(crawlRequest.getUrl()));
222 }
223 continue;
224 }
225 if (!isUrlCrawlable(paramMap, crawlRequest.getUrl())) {
226 continue;
227 }
228 counter++;
229 final Map<String, Object> localDataMap =
230 dataMap.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
231 if (deleteUrlList.contains(crawlRequest.getUrl())) {
232 deleteDocuments();
233 }
234 try {
235 String currentUrl = crawlRequest.getUrl();
236 for (int i = 0; i < maxRedirectCount; i++) {
237 processedUrls.add(currentUrl);
238 if (keyObj != null) {
239 keyObj.setUrl(currentUrl);
240 }
241 crawlerStatsHelper.record(keyObj, StatsAction.PREPARED);
242 currentUrl = processRequest(paramMap, localDataMap, currentUrl, client);
243 if (currentUrl == null) {
244 break;
245 }
246 localDataMap.put(fessConfig.getIndexFieldUrl(), currentUrl);
247 crawlerStatsHelper.record(keyObj, StatsAction.REDIRECTED);
248 }
249 } catch (final ChildUrlsException e) {
250 crawlerStatsHelper.record(keyObj, StatsAction.CHILD_URLS);
251 if (maxDepth == -1 || crawlRequest.getDepth() < maxDepth) {
252 e.getChildUrlList()
253 .stream()
254 .filter(data -> !processedUrls.contains(data.getUrl()))
255 .map(data -> new CrawlRequest(data.getUrl(), crawlRequest.getDepth() + 1))
256 .forEach(requestQueue::offer);
257 }
258 } catch (final DataStoreCrawlingException e) {
259 crawlerStatsHelper.record(keyObj, StatsAction.ACCESS_EXCEPTION);
260 final Throwable cause = e.getCause();
261 if (cause instanceof ChildUrlsException) {
262 if (maxDepth == -1 || crawlRequest.getDepth() < maxDepth) {
263 ((ChildUrlsException) cause).getChildUrlList()
264 .stream()
265 .filter(data -> !processedUrls.contains(data.getUrl()))
266 .map(data -> new CrawlRequest(data.getUrl(), crawlRequest.getDepth() + 1))
267 .forEach(requestQueue::offer);
268 }
269 } else {
270 logger.warn("Failed to access url='{}', depth={}", crawlRequest.getUrl(), crawlRequest.getDepth(), e);
271 }
272 }
273 }
274 }
275 }
276
277
278
279
280
281
282
283
284
285
286
287
288 protected boolean isUrlCrawlable(final DataStoreParams paramMap, final String url) {
289 if (paramMap.containsKey(URL_EXCLUDE_PATTERN)) {
290 if (paramMap.get(URL_EXCLUDE_PATTERN) instanceof final String value) {
291 final Pattern pattern = Pattern.compile(value);
292 paramMap.put(URL_EXCLUDE_PATTERN, pattern);
293 if (logger.isDebugEnabled()) {
294 logger.debug("Using exclude pattern: {}", pattern);
295 }
296 }
297 if (paramMap.get(URL_EXCLUDE_PATTERN) instanceof final Pattern pattern) {
298 if (pattern.matcher(url).matches()) {
299 if (logger.isDebugEnabled()) {
300 logger.debug("Skipping URL {} due to exclude pattern: {}", url, pattern);
301 }
302 return false;
303 }
304 }
305 }
306 return true;
307 }
308
309
310
311
312
313 private static class CrawlRequest {
314
315 private final String url;
316
317 private final int depth;
318
319
320
321
322
323
324
325 CrawlRequest(final String url, final int depth) {
326 this.url = url;
327 this.depth = depth;
328 }
329
330
331
332
333
334
335 public String getUrl() {
336 return url;
337 }
338
339
340
341
342
343
344 public int getDepth() {
345 return depth;
346 }
347
348 @Override
349 public int hashCode() {
350 return Objects.hash(url);
351 }
352
353 @Override
354 public boolean equals(final Object obj) {
355 if (this == obj) {
356 return true;
357 }
358 if (obj == null || getClass() != obj.getClass()) {
359 return false;
360 }
361 final CrawlRequest other = (CrawlRequest) obj;
362 return Objects.equals(url, other.url);
363 }
364
365 @Override
366 public String toString() {
367 return "url=" + url + " depth=" + depth;
368 }
369 }
370
371
372
373
374
375
376
377
378
379 protected long getMaxAccessCount(final DataStoreParams paramMap, final Map<String, Object> dataMap) {
380 if (dataMap.remove(getParamValue(paramMap, "field.max_access_count", "max_access_count")) instanceof final String maxAccessCount
381 && StringUtil.isNotBlank(maxAccessCount)) {
382 try {
383 return Long.parseLong(maxAccessCount);
384 } catch (final NumberFormatException e) {
385 if (logger.isDebugEnabled()) {
386 logger.warn("Failed to parse max_access_count: '{}'. Expected: integer value", maxAccessCount, e);
387 } else {
388 logger.warn("Failed to parse max_access_count: '{}'. Expected: integer value", maxAccessCount);
389 }
390 }
391 }
392
393 final Object recursive = dataMap.remove(getParamValue(paramMap, "field.recursive", "recursive"));
394 if (recursive == null || Constants.FALSE.equalsIgnoreCase(recursive.toString())) {
395 return 1L;
396 }
397 if (Constants.TRUE.equalsIgnoreCase(recursive.toString())) {
398 return -1L;
399 }
400
401 return 1L;
402 }
403
404
405
406
407
408
409
410
411 protected int getMaxDepth(final DataStoreParams paramMap, final Map<String, Object> dataMap) {
412 if (dataMap.remove(getParamValue(paramMap, "field.max_depth", "max_depth")) instanceof final String maxDepth
413 && StringUtil.isNotBlank(maxDepth)) {
414 try {
415 return Integer.parseInt(maxDepth);
416 } catch (final NumberFormatException e) {
417 if (logger.isDebugEnabled()) {
418 logger.warn("Failed to parse max_depth: '{}'. Expected: integer value", maxDepth, e);
419 } else {
420 logger.warn("Failed to parse max_depth: '{}'. Expected: integer value", maxDepth);
421 }
422 }
423 }
424
425 return -1;
426 }
427
428
429
430
431
432
433
434
435
436
437
438
439
440 protected String processRequest(final DataStoreParams paramMap, final Map<String, Object> dataMap, final String url,
441 final CrawlerClient client) {
442 final SystemHelper systemHelper = ComponentUtil.getSystemHelper();
443 final long startTime = systemHelper.getCurrentTimeAsLong();
444 final CrawlerStatsHelper crawlerStatsHelper = ComponentUtil.getCrawlerStatsHelper();
445 final StatsKeyObject keyObj = paramMap.get(Constants.CRAWLER_STATS_KEY) instanceof final StatsKeyObject sko ? sko : null;
446 try (final ResponseData responseData = client.execute(RequestDataBuilder.newRequestData().get().url(url).build())) {
447 if (responseData.getRedirectLocation() != null) {
448 return responseData.getRedirectLocation();
449 }
450 responseData.setExecutionTime(systemHelper.getCurrentTimeAsLong() - startTime);
451 if (dataMap.containsKey(Constants.SESSION_ID)) {
452 responseData.setSessionId((String) dataMap.get(Constants.SESSION_ID));
453 } else {
454 responseData.setSessionId((String) paramMap.get(Constants.CRAWLING_INFO_ID));
455 }
456
457 final RuleManager ruleManager = SingletonLaContainer.getComponent(RuleManager.class);
458 final Rule rule = ruleManager.getRule(responseData);
459 if (rule == null) {
460 logger.warn("No matching url rule for url='{}'", url);
461 } else {
462 responseData.setRuleId(rule.getRuleId());
463 final ResponseProcessor responseProcessor = rule.getResponseProcessor();
464 if (responseProcessor instanceof DefaultResponseProcessor) {
465 final Transformer transformer = ((DefaultResponseProcessor) responseProcessor).getTransformer();
466 final ResultData resultData = transformer.transform(responseData);
467 final Object rawData = resultData.getRawData();
468 if (rawData != null) {
469 @SuppressWarnings("unchecked")
470 final Map<String, Object> responseDataMap = (Map<String, Object>) rawData;
471 mergeResponseData(dataMap, responseDataMap);
472 } else {
473 final byte[] data = resultData.getData();
474 if (data != null) {
475 try {
476 final DataSerializer dataSerializer = ComponentUtil.getComponent("dataSerializer");
477 @SuppressWarnings("unchecked")
478 final Map<String, Object> responseDataMap = (Map<String, Object>) dataSerializer.fromBinaryToObject(data);
479 mergeResponseData(dataMap, responseDataMap);
480 } catch (final Exception e) {
481 throw new CrawlerSystemException("Could not create an instance from bytes.", e);
482 }
483 }
484 }
485 crawlerStatsHelper.record(keyObj, StatsAction.ACCESSED);
486
487
488 String[] ignoreFields;
489 if (paramMap.containsKey("ignore.field.names")) {
490 ignoreFields = ((String) paramMap.get("ignore.field.names")).split(",");
491 } else {
492 ignoreFields = new String[] { Constants.INDEXING_TARGET, Constants.SESSION_ID };
493 }
494 stream(ignoreFields).of(stream -> stream.map(String::trim).forEach(s -> dataMap.remove(s)));
495
496 indexUpdateCallback.store(paramMap, dataMap);
497 crawlerStatsHelper.record(keyObj, StatsAction.PROCESSED);
498 } else {
499 logger.warn("Unexpected response processor: expected=DefaultResponseProcessor, actual={}, url={}",
500 responseProcessor.getClass().getSimpleName(), url);
501 }
502 }
503 return null;
504 } catch (final ChildUrlsException | DataStoreCrawlingException e) {
505 throw e;
506 } catch (final Exception e) {
507 final FessConfig fessConfig = ComponentUtil.getFessConfig();
508 final Object configId = dataMap.get(fessConfig.getIndexFieldConfigId());
509 throw new DataStoreCrawlingException(url, "Failed to add document. url: " + url + ", configId: " + configId, e);
510 }
511 }
512
513
514
515
516
517
518
519
520 protected void mergeResponseData(final Map<String, Object> dataMap, final Map<String, Object> responseDataMap) {
521 dataMap.putAll(responseDataMap);
522 dataMap.keySet()
523 .stream()
524 .filter(key -> key.endsWith(".overwrite"))
525 .collect(Collectors.toList())
526 .forEach(key -> {
527 final String baseKey = key.substring(0, key.length() - ".overwrite".length());
528 final Object value = dataMap.remove(key);
529 dataMap.put(baseKey, value);
530 });
531 }
532
533
534
535
536
537
538
539
540
541
542 protected boolean deleteDocument(final DataStoreParams paramMap, final Map<String, Object> dataMap) {
543
544 final FessConfig fessConfig = ComponentUtil.getFessConfig();
545
546 if (logger.isDebugEnabled()) {
547 logger.debug("Deleting document: url={}", dataMap.get(fessConfig.getIndexFieldUrl()));
548 }
549
550
551 if (!dataMap.containsKey(fessConfig.getIndexFieldUrl()) || dataMap.get(fessConfig.getIndexFieldUrl()) == null) {
552 logger.warn("Could not delete document: url field is missing or null");
553 return false;
554 }
555
556 synchronized (indexUpdateCallback) {
557 final long maxAccessCount = getMaxAccessCount(paramMap, dataMap);
558 final String url = dataMap.get(fessConfig.getIndexFieldUrl()).toString();
559 if (maxAccessCount != 1L) {
560 final SearchEngineClient searchEngineClient = ComponentUtil.getSearchEngineClient();
561 final IndexingHelper indexingHelper = ComponentUtil.getIndexingHelper();
562 final long count = indexingHelper.deleteDocumentByQuery(searchEngineClient,
563 QueryBuilders.prefixQuery(fessConfig.getIndexFieldUrl(), url));
564 if (logger.isDebugEnabled()) {
565 logger.debug("Deleted {} documents for url prefix: {}", count, url);
566 }
567 } else {
568 deleteUrlList.add(url);
569
570 if (deleteUrlList.size() >= maxDeleteDocumentCacheSize) {
571 deleteDocuments();
572 }
573 }
574 }
575 return true;
576 }
577
578 @Override
579 public void commit() {
580 try {
581 if (logger.isDebugEnabled()) {
582 logger.debug("Shutting down thread executor.");
583 }
584 executor.shutdown();
585 executor.awaitTermination(executorTerminationTimeout, TimeUnit.SECONDS);
586 } catch (final InterruptedException e) {
587 if (logger.isDebugEnabled()) {
588 logger.debug("Executor shutdown interrupted", e);
589 }
590 } finally {
591 executor.shutdownNow();
592 }
593
594 synchronized (indexUpdateCallback) {
595 if (!deleteUrlList.isEmpty()) {
596 deleteDocuments();
597 }
598 }
599 indexUpdateCallback.commit();
600 }
601
602
603
604
605
606 protected void deleteDocuments() {
607 final SearchEngineClient searchEngineClient = ComponentUtil.getSearchEngineClient();
608 final IndexingHelper indexingHelper = ComponentUtil.getIndexingHelper();
609 for (final String url : deleteUrlList) {
610 indexingHelper.deleteDocumentByUrl(searchEngineClient, url);
611 }
612 if (logger.isDebugEnabled()) {
613 logger.debug("Deleted {} documents from URL list", deleteUrlList.size());
614 }
615 deleteUrlList.clear();
616 }
617
618 @Override
619 public long getDocumentSize() {
620 return indexUpdateCallback.getDocumentSize();
621 }
622
623 @Override
624 public long getExecuteTime() {
625 return indexUpdateCallback.getExecuteTime();
626 }
627
628
629
630
631
632
633 public void setMaxDeleteDocumentCacheSize(final int maxDeleteDocumentCacheSize) {
634 this.maxDeleteDocumentCacheSize = maxDeleteDocumentCacheSize;
635 }
636
637
638
639
640
641
642 public void setMaxRedirectCount(final int maxRedirectCount) {
643 this.maxRedirectCount = maxRedirectCount;
644 }
645
646
647
648
649
650
651 public void setExecutorTerminationTimeout(final int executorTerminationTimeout) {
652 this.executorTerminationTimeout = executorTerminationTimeout;
653 }
654
655 }