001/* 002 * (C) Copyright 2018 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 * Benoit Delbosc 018 */ 019package org.nuxeo.ecm.automation.elasticsearch; 020 021import static org.nuxeo.elasticsearch.bulk.IndexAction.INDEX_UPDATE_ALIAS_PARAM; 022 023import java.io.IOException; 024import java.util.Collections; 025 026import org.apache.commons.logging.Log; 027import org.apache.commons.logging.LogFactory; 028import org.nuxeo.ecm.automation.OperationContext; 029import org.nuxeo.ecm.automation.core.Constants; 030import org.nuxeo.ecm.automation.core.annotations.Context; 031import org.nuxeo.ecm.automation.core.annotations.Operation; 032import org.nuxeo.ecm.automation.core.annotations.OperationMethod; 033import org.nuxeo.ecm.core.api.Blob; 034import org.nuxeo.ecm.core.api.Blobs; 035import org.nuxeo.ecm.core.api.CoreSession; 036import org.nuxeo.ecm.core.api.NuxeoException; 037import org.nuxeo.ecm.core.api.NuxeoPrincipal; 038import org.nuxeo.ecm.core.bulk.BulkService; 039import org.nuxeo.ecm.core.bulk.message.BulkCommand; 040import org.nuxeo.elasticsearch.api.ElasticSearchAdmin; 041import org.nuxeo.elasticsearch.bulk.IndexAction; 042 043/** 044 * Run Elasticsearch indexing operation using the Bulk Service 045 * 046 * @since 10.3 047 */ 048@Operation(id = ElasticsearchBulkIndexOperation.ID, category = Constants.CAT_SERVICES, label = "Elasticsearch Indexing", since = "10.3", description = "Enable to index Nuxeo documents using the Bulk Service.", addToStudio = false) 049public class ElasticsearchBulkIndexOperation { 050 private static final Log log = LogFactory.getLog(ElasticsearchBulkIndexOperation.class); 051 052 public static final String ID = "Elasticsearch.BulkIndex"; 053 054 @Context 055 protected CoreSession session; 056 057 @Context 058 protected OperationContext ctx; 059 060 @Context 061 protected ElasticSearchAdmin esa; 062 063 @Context 064 protected BulkService bulkService; 065 066 @OperationMethod 067 public Blob run() throws IOException { 068 checkAccess(); 069 esa.dropAndInitRepositoryIndex(session.getRepositoryName(), false); 070 String commandId = submitBulkCommand("SELECT ecm:uuid FROM Document", true); 071 log.warn(String.format("Submitted index command: %s to index the entire %s repository.", commandId, 072 session.getRepositoryName())); 073 return Blobs.createJSONBlobFromValue(Collections.singletonMap("commandId", commandId)); 074 } 075 076 protected String submitBulkCommand(String nxql, boolean syncAlias) { 077 String username = session.getPrincipal().getName(); 078 return bulkService.submit( 079 new BulkCommand.Builder(IndexAction.ACTION_NAME, nxql).param(INDEX_UPDATE_ALIAS_PARAM, syncAlias) 080 .user(username) 081 .build()); 082 } 083 084 protected void checkAccess() { 085 NuxeoPrincipal principal = ctx.getPrincipal(); 086 if (principal == null || !principal.isAdministrator()) { 087 throw new NuxeoException("Unauthorized access: " + principal); 088 } 089 } 090 091 @OperationMethod 092 public Blob run(String nxql) throws IOException { 093 checkAccess(); 094 String commandId = submitBulkCommand(nxql, false); 095 return Blobs.createJSONBlobFromValue(Collections.singletonMap("commandId", commandId)); 096 } 097 098}