001/* 002 * (C) Copyright 2014-2016 Nuxeo SA (http://nuxeo.com/) and others. 003 * 004 * Licensed under the Apache License, Version 2.0 (the "License"); 005 * you may not use this file except in compliance with the License. 006 * You may obtain a copy of the License at 007 * 008 * http://www.apache.org/licenses/LICENSE-2.0 009 * 010 * Unless required by applicable law or agreed to in writing, software 011 * distributed under the License is distributed on an "AS IS" BASIS, 012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 013 * See the License for the specific language governing permissions and 014 * limitations under the License. 015 * 016 * Contributors: 017 * Tiry 018 * Benoit Delbosc 019 */ 020package org.nuxeo.elasticsearch.audit; 021 022import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder; 023import static org.nuxeo.ecm.platform.audit.api.BuiltinLogEntryData.LOG_ID; 024 025import java.io.IOException; 026import java.io.OutputStream; 027import java.io.Serializable; 028import java.util.ArrayList; 029import java.util.Calendar; 030import java.util.Collections; 031import java.util.Date; 032import java.util.HashMap; 033import java.util.Iterator; 034import java.util.List; 035import java.util.Map; 036import java.util.function.Function; 037 038import org.apache.commons.lang3.StringUtils; 039import org.apache.commons.logging.Log; 040import org.apache.commons.logging.LogFactory; 041import org.elasticsearch.action.bulk.BulkItemResponse; 042import org.elasticsearch.action.bulk.BulkRequest; 043import org.elasticsearch.action.bulk.BulkResponse; 044import org.elasticsearch.action.get.GetRequest; 045import org.elasticsearch.action.get.GetResponse; 046import org.elasticsearch.action.index.IndexRequest; 047import org.elasticsearch.action.search.ClearScrollRequest; 048import org.elasticsearch.action.search.SearchRequest; 049import org.elasticsearch.action.search.SearchResponse; 050import org.elasticsearch.action.search.SearchScrollRequest; 051import org.elasticsearch.action.search.SearchType; 052import org.elasticsearch.common.ParsingException; 053import org.elasticsearch.common.io.stream.BytesStreamOutput; 054import org.elasticsearch.common.settings.Settings; 055import org.elasticsearch.common.unit.TimeValue; 056import org.elasticsearch.common.xcontent.NamedXContentRegistry; 057import org.elasticsearch.common.xcontent.XContentBuilder; 058import org.elasticsearch.common.xcontent.XContentFactory; 059import org.elasticsearch.common.xcontent.XContentParser; 060import org.elasticsearch.common.xcontent.XContentType; 061import org.elasticsearch.index.query.BoolQueryBuilder; 062import org.elasticsearch.index.query.QueryBuilder; 063import org.elasticsearch.index.query.QueryBuilders; 064import org.elasticsearch.index.query.QueryParseContext; 065import org.elasticsearch.search.SearchHit; 066import org.elasticsearch.search.SearchModule; 067import org.elasticsearch.search.aggregations.AggregationBuilders; 068import org.elasticsearch.search.aggregations.metrics.max.Max; 069import org.elasticsearch.search.builder.SearchSourceBuilder; 070import org.elasticsearch.search.sort.SortOrder; 071import org.json.JSONException; 072import org.json.JSONObject; 073import org.nuxeo.common.utils.TextTemplate; 074import org.nuxeo.ecm.core.api.CursorResult; 075import org.nuxeo.ecm.core.api.CursorService; 076import org.nuxeo.ecm.core.api.DocumentModel; 077import org.nuxeo.ecm.core.api.NuxeoException; 078import org.nuxeo.ecm.core.api.ScrollResult; 079import org.nuxeo.ecm.core.query.sql.model.Literals; 080import org.nuxeo.ecm.core.query.sql.model.MultiExpression; 081import org.nuxeo.ecm.core.query.sql.model.Operand; 082import org.nuxeo.ecm.core.query.sql.model.Operator; 083import org.nuxeo.ecm.core.query.sql.model.OrderByList; 084import org.nuxeo.ecm.core.query.sql.model.Predicate; 085import org.nuxeo.ecm.core.query.sql.model.Reference; 086import org.nuxeo.ecm.core.uidgen.UIDGeneratorService; 087import org.nuxeo.ecm.core.uidgen.UIDSequencer; 088import org.nuxeo.ecm.core.work.api.Work; 089import org.nuxeo.ecm.core.work.api.Work.State; 090import org.nuxeo.ecm.core.work.api.WorkManager; 091import org.nuxeo.ecm.platform.audit.api.AuditQueryBuilder; 092import org.nuxeo.ecm.platform.audit.api.AuditReader; 093import org.nuxeo.ecm.platform.audit.api.ExtendedInfo; 094import org.nuxeo.ecm.platform.audit.api.LogEntry; 095import org.nuxeo.ecm.platform.audit.impl.LogEntryImpl; 096import org.nuxeo.ecm.platform.audit.service.AbstractAuditBackend; 097import org.nuxeo.ecm.platform.audit.service.AuditBackend; 098import org.nuxeo.ecm.platform.audit.service.BaseLogEntryProvider; 099import org.nuxeo.ecm.platform.audit.service.NXAuditEventsService; 100import org.nuxeo.ecm.platform.audit.service.extension.AuditBackendDescriptor; 101import org.nuxeo.ecm.platform.query.api.PredicateDefinition; 102import org.nuxeo.ecm.platform.query.api.PredicateFieldDefinition; 103import org.nuxeo.elasticsearch.ElasticSearchConstants; 104import org.nuxeo.elasticsearch.api.ESClient; 105import org.nuxeo.elasticsearch.api.ElasticSearchAdmin; 106import org.nuxeo.runtime.api.Framework; 107import org.nuxeo.runtime.model.DefaultComponent; 108 109import com.fasterxml.jackson.core.JsonFactory; 110import com.fasterxml.jackson.core.JsonGenerator; 111import com.fasterxml.jackson.databind.ObjectMapper; 112 113/** 114 * Implementation of the {@link AuditBackend} interface using Elasticsearch persistence 115 * 116 * @author tiry 117 */ 118public class ESAuditBackend extends AbstractAuditBackend implements AuditBackend { 119 120 public static final String SEQ_NAME = "audit"; 121 122 public static final String MIGRATION_FLAG_PROP = "audit.elasticsearch.migration"; 123 124 public static final String MIGRATION_BATCH_SIZE_PROP = "audit.elasticsearch.migration.batchSize"; 125 126 public static final String MIGRATION_DONE_EVENT = "sqlToElasticsearchMigrationDone"; 127 128 public static final int MIGRATION_DEFAULT_BACTH_SIZE = 1000; 129 130 protected CursorService<Iterator<SearchHit>, SearchHit, String> cursorService; 131 132 public ESAuditBackend(NXAuditEventsService component, AuditBackendDescriptor config) { 133 super(component, config); 134 } 135 136 /** 137 * @since 9.3 138 */ 139 public ESAuditBackend() { 140 super(); 141 } 142 143 protected ESClient esClient; 144 145 protected static final Log log = LogFactory.getLog(ESAuditBackend.class); 146 147 protected BaseLogEntryProvider provider = new BaseLogEntryProvider() { 148 149 @Override 150 public int removeEntries(String eventId, String pathPattern) { 151 throw new UnsupportedOperationException("Not implemented yet!"); 152 } 153 154 @Override 155 public void addLogEntry(LogEntry logEntry) { 156 List<LogEntry> entries = new ArrayList<>(); 157 entries.add(logEntry); 158 addLogEntries(entries); 159 } 160 161 }; 162 163 protected ESClient getClient() { 164 log.info("Activate Elasticsearch backend for Audit"); 165 ElasticSearchAdmin esa = Framework.getService(ElasticSearchAdmin.class); 166 ESClient client = esa.getClient(); 167 ensureUIDSequencer(client); 168 return client; 169 } 170 171 protected boolean isMigrationDone() { 172 AuditReader reader = Framework.getService(AuditReader.class); 173 List<LogEntry> entries = reader.queryLogs(new String[] { MIGRATION_DONE_EVENT }, null); 174 return !entries.isEmpty(); 175 } 176 177 @Override 178 public int getApplicationStartedOrder() { 179 int elasticOrder = ((DefaultComponent) Framework.getRuntime().getComponent( 180 "org.nuxeo.elasticsearch.ElasticSearchComponent")).getApplicationStartedOrder(); 181 int uidgenOrder = ((DefaultComponent) Framework.getRuntime().getComponent( 182 "org.nuxeo.ecm.core.uidgen.UIDGeneratorService")).getApplicationStartedOrder(); 183 return Integer.max(elasticOrder, uidgenOrder) + 1; 184 } 185 186 @Override 187 public void onApplicationStarted() { 188 esClient = getClient(); 189 if (Boolean.parseBoolean(Framework.getProperty(MIGRATION_FLAG_PROP))) { 190 if (!isMigrationDone()) { 191 log.info(String.format( 192 "Property %s is true and migration is not done yet, processing audit migration from SQL to Elasticsearch index", 193 MIGRATION_FLAG_PROP)); 194 // Drop audit index first in case of a previous bad migration 195 ElasticSearchAdmin esa = Framework.getService(ElasticSearchAdmin.class); 196 esa.dropAndInitIndex(getESIndexName()); 197 int batchSize = MIGRATION_DEFAULT_BACTH_SIZE; 198 String batchSizeProp = Framework.getProperty(MIGRATION_BATCH_SIZE_PROP); 199 if (batchSizeProp != null) { 200 batchSize = Integer.parseInt(batchSizeProp); 201 } 202 migrate(batchSize); 203 } else { 204 log.warn(String.format( 205 "Property %s is true but migration is already done, please set this property to false", 206 MIGRATION_FLAG_PROP)); 207 } 208 } else { 209 log.debug(String.format("Property %s is false, not processing any migration", MIGRATION_FLAG_PROP)); 210 } 211 cursorService = new CursorService<>(SearchHit::getSourceAsString); 212 } 213 214 @Override 215 public void onApplicationStopped() { 216 if (esClient == null) { 217 return; 218 } 219 try { 220 esClient.close(); 221 cursorService.clear(); 222 } catch (Exception e) { 223 log.warn("Fail to close esClient: " + e.getMessage(), e); 224 } finally { 225 esClient = null; 226 cursorService = null; 227 } 228 } 229 230 @Override 231 @SuppressWarnings("unchecked") 232 public List<LogEntry> queryLogs(AuditQueryBuilder builder) { 233 // prepare parameters 234 Predicate predicate = builder.predicate(); 235 OrderByList orders = builder.orders(); 236 long offset = builder.offset(); 237 long limit = builder.limit(); 238 239 SearchSourceBuilder source = createSearchRequestSource(predicate, orders); 240 241 // Perform search 242 List<LogEntry> logEntries; 243 SearchRequest request = createSearchRequest(); 244 request.source(source); 245 if (limit == 0) { 246 // return all result -> use the scroll api 247 // offset is not taking into account when querying all results 248 TimeValue keepAlive = TimeValue.timeValueMinutes(1); 249 request.scroll(keepAlive); 250 // the size here is the size of each scrolls 251 source.size(100); 252 253 // run request 254 SearchResponse searchResponse = runRequest(request); 255 256 // Build log entries 257 logEntries = buildLogEntries(searchResponse); 258 // Scroll on next results 259 for (; // 260 searchResponse.getHits().getHits().length > 0 261 && logEntries.size() < searchResponse.getHits().getTotalHits(); // 262 searchResponse = runNextScroll(searchResponse.getScrollId(), keepAlive)) { 263 // Build log entries 264 logEntries.addAll(buildLogEntries(searchResponse)); 265 } 266 } else { 267 // return a page -> use a regular search 268 source.from((int) offset).size((int) limit); 269 270 // run request 271 SearchResponse searchResponse = runRequest(request); 272 273 // Build log entries 274 logEntries = buildLogEntries(searchResponse); 275 } 276 277 return logEntries; 278 } 279 280 protected SearchSourceBuilder createSearchRequestSource(Predicate predicate, OrderByList orders) { 281 // create ES query builder 282 QueryBuilder query = createQueryBuilder(predicate); 283 284 // create ES source 285 SearchSourceBuilder source = new SearchSourceBuilder().query(QueryBuilders.constantScoreQuery(query)).size(100); 286 287 // create sort 288 orders.forEach(order -> source.sort(order.reference.name, order.isDescending ? SortOrder.DESC : SortOrder.ASC)); 289 return source; 290 } 291 292 protected QueryBuilder createQueryBuilder(Predicate andPredicate) { 293 // cast parameters 294 // current implementation only support a MultiExpression with AND operator 295 @SuppressWarnings("unchecked") 296 List<Predicate> predicates = (List<Predicate>) ((List<?>) ((MultiExpression) andPredicate).values); 297 // current implementation only use Predicate/OrderByExpr with a simple Reference for left and right 298 Function<Operand, String> getFieldName = operand -> ((Reference) operand).name; 299 300 QueryBuilder query; 301 if (predicates.isEmpty()) { 302 query = QueryBuilders.matchAllQuery(); 303 } else { 304 BoolQueryBuilder boolQuery = QueryBuilders.boolQuery(); 305 for (Predicate predicate : predicates) { 306 String leftName = getFieldName.apply(predicate.lvalue); 307 Operator operator = predicate.operator; 308 Object rightValue = Literals.valueOf(predicate.rvalue); 309 if (Operator.EQ.equals(operator)) { 310 boolQuery.must(QueryBuilders.termQuery(leftName, rightValue)); 311 } else if (Operator.NOTEQ.equals(operator)) { 312 boolQuery.mustNot(QueryBuilders.termQuery(leftName, rightValue)); 313 } else if (Operator.LT.equals(operator)) { 314 boolQuery.must(QueryBuilders.rangeQuery(leftName).lt(rightValue)); 315 } else if (Operator.LTEQ.equals(operator)) { 316 boolQuery.must(QueryBuilders.rangeQuery(leftName).lte(rightValue)); 317 } else if (Operator.GTEQ.equals(operator)) { 318 boolQuery.must(QueryBuilders.rangeQuery(leftName).gte(rightValue)); 319 } else if (Operator.GT.equals(operator)) { 320 boolQuery.must(QueryBuilders.rangeQuery(leftName).gt(rightValue)); 321 } else if (Operator.IN.equals(operator)) { 322 boolQuery.must(QueryBuilders.termsQuery(leftName, (List<?>) rightValue)); 323 } 324 } 325 query = boolQuery; 326 } 327 return query; 328 } 329 330 protected List<LogEntry> buildLogEntries(SearchResponse searchResponse) { 331 List<LogEntry> entries = new ArrayList<>(searchResponse.getHits().getHits().length); 332 ObjectMapper mapper = new ObjectMapper(); 333 for (SearchHit hit : searchResponse.getHits()) { 334 try { 335 entries.add(mapper.readValue(hit.getSourceAsString(), LogEntryImpl.class)); 336 } catch (IOException e) { 337 log.error("Error while reading Audit Entry from ES", e); 338 } 339 } 340 return entries; 341 } 342 343 protected SearchRequest createSearchRequest() { 344 return new SearchRequest(getESIndexName()).types(ElasticSearchConstants.ENTRY_TYPE) 345 .searchType(SearchType.DFS_QUERY_THEN_FETCH); 346 } 347 348 @Override 349 public LogEntry getLogEntryByID(long id) { 350 GetResponse ret = esClient.get( 351 new GetRequest(getESIndexName(), ElasticSearchConstants.ENTRY_TYPE, String.valueOf(id))); 352 if (!ret.isExists()) { 353 return null; 354 } 355 try { 356 return new ObjectMapper().readValue(ret.getSourceAsString(), LogEntryImpl.class); 357 } catch (IOException e) { 358 throw new NuxeoException("Unable to read Entry for id " + id, e); 359 } 360 } 361 362 public SearchRequest buildQuery(String query, Map<String, Object> params) { 363 if (params != null && params.size() > 0) { 364 query = expandQueryVariables(query, params); 365 } 366 SearchRequest request = createSearchRequest(); 367 SearchSourceBuilder sourceBuilder = createSearchSourceBuilder(query); 368 return request.source(sourceBuilder); 369 } 370 371 protected SearchSourceBuilder createSearchSourceBuilder(String query) { 372 SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder(); 373 SearchModule searchModule = new SearchModule(Settings.EMPTY, false, Collections.emptyList()); 374 try { 375 try (XContentParser parser = XContentFactory.xContent(XContentType.JSON).createParser( 376 new NamedXContentRegistry(searchModule.getNamedXContents()), query)) { 377 searchSourceBuilder.parseXContent(new QueryParseContext(parser)); 378 } 379 } catch (IOException | ParsingException e) { 380 log.error("Invalid query: " + query + ": " + e.getMessage(), e); 381 throw new IllegalArgumentException("Bad query: " + query); 382 } 383 return searchSourceBuilder; 384 } 385 386 public String expandQueryVariables(String query, Object[] params) { 387 Map<String, Object> qParams = new HashMap<>(); 388 for (int i = 0; i < params.length; i++) { 389 query = query.replaceFirst("\\?", "\\${param" + i + "}"); 390 qParams.put("param" + i, params[i]); 391 } 392 return expandQueryVariables(query, qParams); 393 } 394 395 public String expandQueryVariables(String query, Map<String, Object> params) { 396 if (params != null && params.size() > 0) { 397 TextTemplate tmpl = new TextTemplate(); 398 for (String key : params.keySet()) { 399 Object val = params.get(key); 400 if (val == null) { 401 continue; 402 } else if (val instanceof Calendar) { 403 tmpl.setVariable(key, Long.toString(((Calendar) val).getTime().getTime())); 404 } else if (val instanceof Date) { 405 tmpl.setVariable(key, Long.toString(((Date) val).getTime())); 406 } else { 407 tmpl.setVariable(key, val.toString()); 408 } 409 } 410 query = tmpl.processText(query); 411 } 412 return query; 413 } 414 415 @Override 416 public List<?> nativeQuery(String query, Map<String, Object> params, int pageNb, int pageSize) { 417 SearchRequest request = buildQuery(query, params); 418 if (pageNb > 0) { 419 request.source().from(pageNb * pageSize); 420 } 421 if (pageSize > 0) { 422 request.source().size(pageSize); 423 } 424 SearchResponse searchResponse = runRequest(request); 425 return buildLogEntries(searchResponse); 426 } 427 428 @Override 429 public List<LogEntry> queryLogsByPage(String[] eventIds, Date limit, String[] categories, String path, int pageNb, 430 int pageSize) { 431 SearchRequest request = createSearchRequest(); 432 BoolQueryBuilder filterBuilder = QueryBuilders.boolQuery(); 433 if (eventIds != null && eventIds.length > 0) { 434 if (eventIds.length == 1) { 435 filterBuilder.must(QueryBuilders.termQuery("eventId", eventIds[0])); 436 } else { 437 filterBuilder.must(QueryBuilders.termsQuery("eventId", eventIds)); 438 } 439 } 440 if (categories != null && categories.length > 0) { 441 if (categories.length == 1) { 442 filterBuilder.must(QueryBuilders.termQuery("category", categories[0])); 443 } else { 444 filterBuilder.must(QueryBuilders.termsQuery("category", categories)); 445 } 446 } 447 if (path != null) { 448 filterBuilder.must(QueryBuilders.termQuery("docPath", path)); 449 } 450 451 if (limit != null) { 452 filterBuilder.must(QueryBuilders.rangeQuery("eventDate").lt(convertDate(limit))); 453 } 454 request.source(new SearchSourceBuilder().query(QueryBuilders.constantScoreQuery(filterBuilder))); 455 if (pageNb > 0) { 456 request.source().from(pageNb * pageSize); 457 } 458 if (pageSize > 0) { 459 request.source().size(pageSize); 460 } 461 SearchResponse searchResponse = runRequest(request); 462 return buildLogEntries(searchResponse); 463 } 464 465 @Override 466 public void addLogEntries(List<LogEntry> entries) { 467 468 if (entries.isEmpty()) { 469 return; 470 } 471 472 BulkRequest bulkRequest = new BulkRequest(); 473 JsonFactory factory = new JsonFactory(); 474 475 UIDGeneratorService uidGeneratorService = Framework.getService(UIDGeneratorService.class); 476 UIDSequencer seq = uidGeneratorService.getSequencer(); 477 478 try { 479 480 for (LogEntry entry : entries) { 481 entry.setId(seq.getNextLong(SEQ_NAME)); 482 if (log.isDebugEnabled()) { 483 log.debug(String.format("Indexing log entry: %s", entry)); 484 } 485 entry.setLogDate(new Date()); 486 OutputStream out = new BytesStreamOutput(); 487 JsonGenerator jsonGen = factory.createGenerator(out); 488 XContentBuilder builder = jsonBuilder(out); 489 ObjectMapper mapper = new ObjectMapper(); 490 mapper.writeValue(jsonGen, entry); 491 bulkRequest.add(new IndexRequest(getESIndexName(), ElasticSearchConstants.ENTRY_TYPE, 492 String.valueOf(entry.getId())).source(builder)); 493 } 494 495 BulkResponse bulkResponse = esClient.bulk(bulkRequest); 496 if (bulkResponse.hasFailures()) { 497 for (BulkItemResponse response : bulkResponse.getItems()) { 498 if (response.isFailed()) { 499 log.error("Unable to index audit entry " + response.getItemId() + " :" 500 + response.getFailureMessage()); 501 } 502 } 503 } 504 } catch (IOException e) { 505 throw new NuxeoException("Error while indexing Audit entries", e); 506 } 507 508 } 509 510 @Override 511 public Long getEventsCount(String eventId) { 512 SearchResponse res = esClient.search( 513 new SearchRequest(getESIndexName()).types(ElasticSearchConstants.ENTRY_TYPE).source( 514 new SearchSourceBuilder().query( 515 QueryBuilders.constantScoreQuery(QueryBuilders.termQuery("eventId", eventId))) 516 .size(0))); 517 return res.getHits().getTotalHits(); 518 } 519 520 @Override 521 public long syncLogCreationEntries(final String repoId, final String path, final Boolean recurs) { 522 return syncLogCreationEntries(provider, repoId, path, recurs); 523 } 524 525 public SearchResponse search(SearchRequest request) { 526 String[] indices = request.indices(); 527 if (indices == null || indices.length != 1) { 528 throw new IllegalStateException("Search on audit must include index name: " + request); 529 } 530 if (!getESIndexName().equals(indices[0])) { 531 throw new IllegalStateException("Search on audit must be on audit index: " + request); 532 } 533 return runRequest(request); 534 } 535 536 protected QueryBuilder buildFilter(PredicateDefinition[] predicates, DocumentModel searchDocumentModel) { 537 538 if (searchDocumentModel == null) { 539 return QueryBuilders.matchAllQuery(); 540 } 541 542 BoolQueryBuilder filterBuilder = QueryBuilders.boolQuery(); 543 544 int nbFilters = 0; 545 546 for (PredicateDefinition predicate : predicates) { 547 548 // extract data from DocumentModel 549 PredicateFieldDefinition[] fieldDef = predicate.getValues(); 550 Object[] val = new Object[fieldDef.length]; 551 for (int fidx = 0; fidx < fieldDef.length; fidx++) { 552 if (fieldDef[fidx].getXpath() != null) { 553 val[fidx] = searchDocumentModel.getPropertyValue(fieldDef[fidx].getXpath()); 554 } else { 555 val[fidx] = searchDocumentModel.getProperty(fieldDef[fidx].getSchema(), fieldDef[fidx].getName()); 556 } 557 } 558 559 if (!isNonNullParam(val)) { 560 // skip predicate where all values are null 561 continue; 562 } 563 564 nbFilters++; 565 566 String op = predicate.getOperator(); 567 if (op.equalsIgnoreCase("IN")) { 568 569 String[] values = null; 570 if (val[0] instanceof Iterable<?>) { 571 List<String> l = new ArrayList<>(); 572 Iterable<?> vals = (Iterable<?>) val[0]; 573 574 for (Object v : vals) { 575 if (v != null) { 576 l.add(v.toString()); 577 } 578 } 579 values = l.toArray(new String[l.size()]); 580 } else if (val[0] instanceof Object[]) { 581 values = (String[]) val[0]; 582 } 583 filterBuilder.must(QueryBuilders.termsQuery(predicate.getParameter(), values)); 584 } else if (op.equalsIgnoreCase("BETWEEN")) { 585 filterBuilder.must(QueryBuilders.rangeQuery(predicate.getParameter()).gt(convertDate(val[0]))); 586 if (val.length > 1) { 587 filterBuilder.must(QueryBuilders.rangeQuery(predicate.getParameter()).lt(convertDate(val[1]))); 588 } 589 } else if (">".equals(op)) { 590 filterBuilder.must(QueryBuilders.rangeQuery(predicate.getParameter()).gt(convertDate(val[0]))); 591 } else if (">=".equals(op)) { 592 filterBuilder.must(QueryBuilders.rangeQuery(predicate.getParameter()).gte(convertDate(val[0]))); 593 } else if ("<".equals(op)) { 594 filterBuilder.must(QueryBuilders.rangeQuery(predicate.getParameter()).lt(convertDate(val[0]))); 595 } else if ("<=".equals(op)) { 596 filterBuilder.must(QueryBuilders.rangeQuery(predicate.getParameter()).lte(convertDate(val[0]))); 597 } else { 598 filterBuilder.must(QueryBuilders.termQuery(predicate.getParameter(), convertDate(val[0]))); 599 } 600 } 601 602 if (nbFilters == 0) { 603 return QueryBuilders.matchAllQuery(); 604 } 605 return filterBuilder; 606 } 607 608 protected Object convertDate(Object o) { 609 // Date are convert to timestamp ms which is a known format by default for ES 610 if (o instanceof Calendar) { 611 return Long.valueOf(((Calendar) o).getTime().getTime()); 612 } else if (o instanceof Date) { 613 return Long.valueOf(((Date) o).getTime()); 614 } 615 return o; 616 } 617 618 public SearchRequest buildSearchQuery(String fixedPart, PredicateDefinition[] predicates, 619 DocumentModel searchDocumentModel) { 620 SearchRequest request = createSearchRequest(); 621 QueryBuilder queryBuilder = QueryBuilders.wrapperQuery(fixedPart); 622 QueryBuilder filterBuilder = buildFilter(predicates, searchDocumentModel); 623 request.source( 624 new SearchSourceBuilder().query(QueryBuilders.boolQuery().must(queryBuilder).filter(filterBuilder))); 625 return request; 626 } 627 628 protected boolean isNonNullParam(Object[] val) { 629 if (val == null) { 630 return false; 631 } 632 for (Object v : val) { 633 if (v != null) { 634 if (v instanceof String) { 635 if (!((String) v).isEmpty()) { 636 return true; 637 } 638 } else if (v instanceof String[]) { 639 if (((String[]) v).length > 0) { 640 return true; 641 } 642 } else { 643 return true; 644 } 645 } 646 } 647 return false; 648 } 649 650 @SuppressWarnings("deprecation") 651 public String migrate(final int batchSize) { 652 653 final String MIGRATION_WORK_ID = "AuditMigration"; 654 655 WorkManager wm = Framework.getService(WorkManager.class); 656 State migrationState = wm.getWorkState(MIGRATION_WORK_ID); 657 if (migrationState != null) { 658 return "Migration already scheduled : " + migrationState.toString(); 659 } 660 661 Work migrationWork = new ESAuditMigrationWork(MIGRATION_WORK_ID, batchSize); 662 wm.schedule(migrationWork); 663 return "Migration work started : " + MIGRATION_WORK_ID; 664 } 665 666 SearchResponse runRequest(SearchRequest request) { 667 logSearchRequest(request); 668 SearchResponse response = esClient.search(request); 669 logSearchResponse(response); 670 return response; 671 } 672 673 SearchResponse runNextScroll(String scrollId, TimeValue keepAlive) { 674 if (log.isDebugEnabled()) { 675 log.debug(String.format( 676 "Scroll request: -XGET 'localhost:9200/_search/scroll' -d '{\"scroll\": \"%s\", \"scroll_id\": \"%s\" }'", 677 keepAlive, scrollId)); 678 } 679 SearchResponse response = esClient.searchScroll(new SearchScrollRequest(scrollId).scroll(keepAlive)); 680 logSearchResponse(response); 681 return response; 682 } 683 684 protected void logSearchResponse(SearchResponse response) { 685 if (log.isDebugEnabled()) { 686 log.debug("Response: " + response.toString()); 687 } 688 } 689 690 protected void logSearchRequest(SearchRequest request) { 691 if (log.isDebugEnabled()) { 692 log.debug(String.format("Search query: curl -XGET 'http://localhost:9200/%s/%s/_search?pretty' -d '%s'", 693 getESIndexName(), ElasticSearchConstants.ENTRY_TYPE, request.toString())); 694 } 695 } 696 697 /** 698 * Ensures the audit sequence returns an UID greater or equal than the maximum log entry id. 699 */ 700 protected void ensureUIDSequencer(ESClient esClient) { 701 boolean auditIndexExists = esClient.indexExists(getESIndexName()); 702 if (!auditIndexExists) { 703 return; 704 } 705 706 // Get max log entry id 707 SearchRequest request = createSearchRequest(); 708 request.source(new SearchSourceBuilder().query(QueryBuilders.matchAllQuery()) 709 .aggregation(AggregationBuilders.max("maxAgg").field("id"))); 710 SearchResponse searchResponse = esClient.search(request); 711 Max agg = searchResponse.getAggregations().get("maxAgg"); 712 long maxLogEntryId = (long) agg.getValue(); 713 714 // Get next sequence id 715 UIDGeneratorService uidGeneratorService = Framework.getService(UIDGeneratorService.class); 716 UIDSequencer seq = uidGeneratorService.getSequencer(); 717 seq.init(); 718 long nextSequenceId = seq.getNextLong(SEQ_NAME); 719 720 // Increment sequence to max log entry id if needed 721 if (nextSequenceId < maxLogEntryId) { 722 log.info(String.format("Next UID returned by %s sequence is %d, initializing sequence to %d", SEQ_NAME, 723 nextSequenceId, maxLogEntryId)); 724 seq.initSequence(SEQ_NAME, maxLogEntryId); 725 } 726 } 727 728 @Override 729 public ExtendedInfo newExtendedInfo(Serializable value) { 730 return new ESExtendedInfo(value); 731 } 732 733 protected String getESIndexName() { 734 ElasticSearchAdmin esa = Framework.getService(ElasticSearchAdmin.class); 735 return esa.getIndexNameForType(ElasticSearchConstants.ENTRY_TYPE); 736 } 737 738 @Override 739 public void append(List<String> jsonEntries) { 740 BulkRequest bulkRequest = new BulkRequest(); 741 for (String json : jsonEntries) { 742 try { 743 String entryId = new JSONObject(json).getString(LOG_ID); 744 if (StringUtils.isBlank(entryId)) { 745 throw new NuxeoException("A json entry has an empty id. entry=" + json); 746 } 747 IndexRequest request = new IndexRequest(getESIndexName(), ElasticSearchConstants.ENTRY_TYPE, entryId); 748 request.source(json, XContentType.JSON); 749 bulkRequest.add(request); 750 } catch (JSONException e) { 751 throw new NuxeoException("Unable to deserialize json entry=" + json, e); 752 } 753 } 754 esClient.bulk(bulkRequest); 755 } 756 757 @Override 758 public ScrollResult<String> scroll(AuditQueryBuilder builder, int batchSize, int keepAliveSeconds) { 759 // prepare parameters 760 Predicate predicate = builder.predicate(); 761 OrderByList orders = builder.orders(); 762 763 // create source 764 SearchSourceBuilder source = createSearchRequestSource(predicate, orders); 765 source.size(batchSize); 766 // create request 767 SearchRequest request = createSearchRequest(); 768 request.source(source).scroll(TimeValue.timeValueSeconds(keepAliveSeconds)); 769 SearchResponse response = runRequest(request); 770 // register cursor 771 String scrollId = cursorService.registerCursorResult(new ESCursorResult(response, batchSize, keepAliveSeconds)); 772 return scroll(scrollId); 773 } 774 775 @Override 776 public ScrollResult<String> scroll(String scrollId) { 777 return cursorService.scroll(scrollId); 778 } 779 780 public class ESCursorResult extends CursorResult<Iterator<SearchHit>, SearchHit> { 781 782 protected final String scrollId; 783 784 protected boolean end; 785 786 public ESCursorResult(SearchResponse response, int batchSize, int keepAliveSeconds) { 787 super(response.getHits().iterator(), batchSize, keepAliveSeconds); 788 this.scrollId = response.getScrollId(); 789 } 790 791 @Override 792 public boolean hasNext() { 793 if (cursor == null || end) { 794 return false; 795 } else if (cursor.hasNext()) { 796 return true; 797 } else { 798 runNextScroll(); 799 return !end; 800 } 801 } 802 803 @Override 804 public SearchHit next() { 805 if (cursor != null && !cursor.hasNext() && !end) { 806 // try to run a next scroll 807 runNextScroll(); 808 } 809 return super.next(); 810 } 811 812 protected void runNextScroll() { 813 SearchResponse response = ESAuditBackend.this.runNextScroll(scrollId, 814 TimeValue.timeValueSeconds(keepAliveSeconds)); 815 cursor = response.getHits().iterator(); 816 end = !cursor.hasNext(); 817 } 818 819 @Override 820 public void close() { 821 ClearScrollRequest request = new ClearScrollRequest(); 822 request.addScrollId(scrollId); 823 esClient.clearScroll(request); 824 end = true; 825 // Call super close to clear cursor 826 super.close(); 827 } 828 829 } 830 831}