1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package org.codelibs.fess.ds.impl;
17
18 import static org.codelibs.core.stream.StreamUtil.split;
19
20 import java.net.InetAddress;
21 import java.util.HashMap;
22 import java.util.LinkedHashMap;
23 import java.util.Map;
24 import java.util.stream.Collectors;
25
26 import org.codelibs.core.lang.StringUtil;
27 import org.codelibs.fess.Constants;
28 import org.codelibs.fess.app.service.FailureUrlService;
29 import org.codelibs.fess.crawler.exception.CrawlingAccessException;
30 import org.codelibs.fess.crawler.exception.MultipleCrawlingAccessException;
31 import org.codelibs.fess.ds.IndexUpdateCallback;
32 import org.codelibs.fess.es.config.exentity.DataConfig;
33 import org.codelibs.fess.exception.DataStoreCrawlingException;
34 import org.codelibs.fess.exception.DataStoreException;
35 import org.codelibs.fess.util.ComponentUtil;
36 import org.elasticsearch.action.bulk.BulkRequestBuilder;
37 import org.elasticsearch.action.bulk.BulkResponse;
38 import org.elasticsearch.action.search.SearchRequestBuilder;
39 import org.elasticsearch.action.search.SearchResponse;
40 import org.elasticsearch.client.Client;
41 import org.elasticsearch.common.settings.Settings;
42 import org.elasticsearch.common.transport.InetSocketTransportAddress;
43 import org.elasticsearch.index.query.QueryBuilders;
44 import org.elasticsearch.search.SearchHit;
45 import org.elasticsearch.search.SearchHits;
46 import org.elasticsearch.transport.client.PreBuiltTransportClient;
47 import org.slf4j.Logger;
48 import org.slf4j.LoggerFactory;
49
50 public class EsDataStoreImpl extends AbstractDataStoreImpl {
51 private static final String PREFERENCE = "preference";
52
53 private static final String QUERY = "query";
54
55 private static final String FIELDS = "fields";
56
57 private static final String SIZE = "size";
58
59 private static final String TYPE = "type";
60
61 private static final String TIMEOUT = "timeout";
62
63 private static final String SCROLL = "scroll";
64
65 private static final String INDEX = "index";
66
67 private static final String HOSTS = "hosts";
68
69 private static final String SETTINGS_PREFIX = "settings.";
70
71 private static final Logger logger = LoggerFactory.getLogger(EsDataStoreImpl.class);
72
73 @Override
74 protected void storeData(final DataConfig dataConfig, final IndexUpdateCallback callback, final Map<String, String> paramMap,
75 final Map<String, String> scriptMap, final Map<String, Object> defaultDataMap) {
76 final String hostsStr = paramMap.get(HOSTS);
77 if (StringUtil.isBlank(hostsStr)) {
78 logger.info("hosts is empty.");
79 return;
80 }
81
82 final long readInterval = getReadInterval(paramMap);
83
84 final Settings settings =
85 Settings.builder()
86 .put(paramMap
87 .entrySet()
88 .stream()
89 .filter(e -> e.getKey().startsWith(SETTINGS_PREFIX))
90 .collect(
91 Collectors.toMap(e -> e.getKey().replaceFirst("^settings\\.", StringUtil.EMPTY), e -> e.getValue())))
92 .build();
93 logger.info("Connecting to " + hostsStr + " with [" + settings.toDelimitedString(',') + "]");
94 final InetSocketTransportAddress[] addresses = split(hostsStr, ",").get(stream -> stream.map(h -> {
95 final String[] values = h.trim().split(":");
96 try {
97 if (values.length == 1) {
98 return new InetSocketTransportAddress(InetAddress.getByName(values[0]), 9300);
99 } else if (values.length == 2) {
100 return new InetSocketTransportAddress(InetAddress.getByName(values[0]), Integer.parseInt(values[1]));
101 }
102 } catch (final Exception e) {
103 logger.warn("Failed to parse address: " + h, e);
104 }
105 return null;
106 }).filter(v -> v != null).toArray(n -> new InetSocketTransportAddress[n]));
107 try (PreBuiltTransportClient client = new PreBuiltTransportClient(settings)) {
108 client.addTransportAddresses(addresses);
109 processData(dataConfig, callback, paramMap, scriptMap, defaultDataMap, readInterval, client);
110 }
111 }
112
113 protected void processData(final DataConfig dataConfig, final IndexUpdateCallback callback, final Map<String, String> paramMap,
114 final Map<String, String> scriptMap, final Map<String, Object> defaultDataMap, final long readInterval, final Client client) {
115
116 final boolean deleteProcessedDoc = paramMap.getOrDefault("delete.processed.doc", Constants.FALSE).equalsIgnoreCase(Constants.TRUE);
117 final String[] indices;
118 if (paramMap.containsKey(INDEX)) {
119 indices = paramMap.get(INDEX).trim().split(",");
120 } else {
121 indices = new String[] { "_all" };
122 }
123 final String scroll = paramMap.containsKey(SCROLL) ? paramMap.get(SCROLL).trim() : "1m";
124 final String timeout = paramMap.containsKey(TIMEOUT) ? paramMap.get(TIMEOUT).trim() : "1m";
125 final SearchRequestBuilder builder = client.prepareSearch(indices);
126 if (paramMap.containsKey(TYPE)) {
127 builder.setTypes(paramMap.get(TYPE).trim().split(","));
128 }
129 if (paramMap.containsKey(SIZE)) {
130 builder.setSize(Integer.parseInt(paramMap.get(SIZE)));
131 }
132 if (paramMap.containsKey(FIELDS)) {
133 builder.setFetchSource(paramMap.get(FIELDS).trim().split(","), null);
134 }
135 builder.setQuery(QueryBuilders.wrapperQuery(paramMap.containsKey(QUERY) ? paramMap.get(QUERY).trim() : "{\"match_all\":{}}"));
136 builder.setScroll(scroll);
137 builder.setPreference(paramMap.containsKey(PREFERENCE) ? paramMap.get(PREFERENCE).trim() : Constants.SEARCH_PREFERENCE_PRIMARY);
138 try {
139 SearchResponse response = builder.execute().actionGet(timeout);
140
141 String scrollId = response.getScrollId();
142 while (scrollId != null) {
143 final SearchHits searchHits = response.getHits();
144 final SearchHit[] hits = searchHits.getHits();
145 if (hits.length == 0) {
146 scrollId = null;
147 break;
148 }
149
150 boolean loop = true;
151 final BulkRequestBuilder bulkRequest = deleteProcessedDoc ? client.prepareBulk() : null;
152 for (final SearchHit hit : hits) {
153 if (!alive || !loop) {
154 break;
155 }
156
157 final Map<String, Object> dataMap = new HashMap<>();
158 dataMap.putAll(defaultDataMap);
159 final Map<String, Object> resultMap = new LinkedHashMap<>();
160 resultMap.putAll(paramMap);
161 resultMap.put("index", hit.getIndex());
162 resultMap.put("type", hit.getType());
163 resultMap.put("id", hit.getId());
164 resultMap.put("version", Long.valueOf(hit.getVersion()));
165 resultMap.put("hit", hit);
166 resultMap.put("source", hit.getSource());
167 resultMap.put("crawlingConfig", dataConfig);
168
169 if (logger.isDebugEnabled()) {
170 for (final Map.Entry<String, Object> entry : resultMap.entrySet()) {
171 logger.debug(entry.getKey() + "=" + entry.getValue());
172 }
173 }
174
175 final Map<String, Object> crawlingContext = new HashMap<>();
176 crawlingContext.put("doc", dataMap);
177 resultMap.put("crawlingContext", crawlingContext);
178 for (final Map.Entry<String, String> entry : scriptMap.entrySet()) {
179 final Object convertValue = convertValue(entry.getValue(), resultMap);
180 if (convertValue != null) {
181 dataMap.put(entry.getKey(), convertValue);
182 }
183 }
184
185 if (logger.isDebugEnabled()) {
186 for (final Map.Entry<String, Object> entry : dataMap.entrySet()) {
187 logger.debug(entry.getKey() + "=" + entry.getValue());
188 }
189 }
190
191 try {
192 callback.store(paramMap, dataMap);
193 } catch (final CrawlingAccessException e) {
194 logger.warn("Crawling Access Exception at : " + dataMap, e);
195
196 Throwable target = e;
197 if (target instanceof MultipleCrawlingAccessException) {
198 final Throwable[] causes = ((MultipleCrawlingAccessException) target).getCauses();
199 if (causes.length > 0) {
200 target = causes[causes.length - 1];
201 }
202 }
203
204 String errorName;
205 final Throwable cause = target.getCause();
206 if (cause != null) {
207 errorName = cause.getClass().getCanonicalName();
208 } else {
209 errorName = target.getClass().getCanonicalName();
210 }
211
212 String url;
213 if (target instanceof DataStoreCrawlingException) {
214 final DataStoreCrawlingException dce = (DataStoreCrawlingException) target;
215 url = dce.getUrl();
216 if (dce.aborted()) {
217 loop = false;
218 }
219 } else {
220 url = hit.getIndex() + "/" + hit.getType() + "/" + hit.getId();
221 }
222 final FailureUrlService failureUrlService = ComponentUtil.getComponent(FailureUrlService.class);
223 failureUrlService.store(dataConfig, errorName, url, target);
224 } catch (final Throwable t) {
225 logger.warn("Crawling Access Exception at : " + dataMap, t);
226 final String url = hit.getIndex() + "/" + hit.getType() + "/" + hit.getId();
227 final FailureUrlService failureUrlService = ComponentUtil.getComponent(FailureUrlService.class);
228 failureUrlService.store(dataConfig, t.getClass().getCanonicalName(), url, t);
229 }
230
231 if (bulkRequest != null) {
232 bulkRequest.add(client.prepareDelete(hit.getIndex(), hit.getType(), hit.getId()));
233 }
234
235 if (readInterval > 0) {
236 sleep(readInterval);
237 }
238 }
239
240 if (bulkRequest != null && bulkRequest.numberOfActions() > 0) {
241 final BulkResponse bulkResponse = bulkRequest.execute().actionGet(timeout);
242 if (bulkResponse.hasFailures()) {
243 logger.warn(bulkResponse.buildFailureMessage());
244 }
245 }
246
247 if (!alive) {
248 break;
249 }
250 response = client.prepareSearchScroll(scrollId).setScroll(scroll).execute().actionGet(timeout);
251 scrollId = response.getScrollId();
252 }
253 } catch (final Exception e) {
254 throw new DataStoreException("Failed to crawl data when acessing elasticsearch.", e);
255 }
256 }
257
258 }