<commit_before>/**
f_ * Copyright 2016, RadiantBlue Technologies, Inc.
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 **/
package access.messaging;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Future;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.Producer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Component;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;

import access.database.DatabaseAccessor;
import access.deploy.Deployer;
import access.deploy.GroupDeployer;
import access.deploy.Leaser;
import exception.DataInspectException;
import exception.GeoServerException;
import exception.InvalidInputException;
import messaging.job.JobMessageFactory;
import messaging.job.WorkerCallback;
import model.data.DataResource;
import model.data.deployment.Deployment;
import model.data.deployment.DeploymentGroup;
import model.job.Job;
import model.job.result.type.DeploymentResult;
import model.job.result.type.ErrorResult;
import model.job.type.AccessJob;
import model.logger.AuditElement;
import model.logger.Severity;
import model.status.StatusUpdate;
import util.PiazzaLogger;

/**
 * Worker class that handles Access Jobs being passed in through the Kafka. Handles the Access Jobs by standing up
 * services, retrieving files, or other various forms of Access for data.
 * 
 * This component assumes that the data intended to be accessed is already ingested into the Piazza system; either by
 * the Ingest component or other components that are capable of inserting data into Piazza.
 * 
 * @author Patrick.Doody & Sonny.Saniev
 * 
 */
@Component
public class AccessWorker {
	@Autowired
	private Deployer deployer;
	@Autowired
	private GroupDeployer groupDeployer;
	@Autowired
	private DatabaseAccessor accessor;
	@Autowired
	private Leaser leaser;
	@Autowired
	private PiazzaLogger pzLogger;
	@Value("${SPACE}")
	private String space;

	private static final Logger LOGGER = LoggerFactory.getLogger(AccessWorker.class);

	/**
	 * Listens for Kafka Access messages for creating Deployments for Access of Resources
	 */
	@Async
	public Future<AccessJob> run(ConsumerRecord<String, String> consumerRecord, Producer<String, String> producer,
			WorkerCallback callback) {
		AccessJob accessJob = null;
		try {
			// Parse Job information from Kafka
			ObjectMapper mapper = new ObjectMapper();
			Job job = mapper.readValue(consumerRecord.value(), Job.class);
			accessJob = (AccessJob) job.getJobType();

			// Validate inputs for the Kafka Message
			if ((accessJob.getDataId() == null) || (accessJob.getDataId().isEmpty())) {
				throw new InvalidInputException(String.format("An invalid or empty Data Id was specified: %s", accessJob.getDataId()));
			}

			if ((accessJob.getDeploymentType() == null) || (accessJob.getDeploymentType().isEmpty())) {
				throw new InvalidInputException(
						String.format("An invalid or empty Deployment Type was specified: %s", accessJob.getDataId()));
			}

			// Logging
			pzLogger.log(
					String.format("Received Request to Access Data %s of Type %s under Job Id %s by user %s", accessJob.getDataId(),
							accessJob.getDeploymentType(), job.getJobId(), job.getCreatedBy()),
					Severity.INFORMATIONAL, new AuditElement(job.getJobId(), "requestAccessData", accessJob.getDataId()));

			if (Thread.interrupted()) {
				throw new InterruptedException();
			}

			processGeoServerType(job, accessJob, producer, consumerRecord.key());

		} catch (InterruptedException exception) {
			String error = String.format("Thread interrupt received for Job %s", consumerRecord.key());
			LOGGER.error(error, exception);
			pzLogger.log(error, Severity.INFORMATIONAL, new AuditElement(consumerRecord.key(), "accessJobTerminated", ""));
			StatusUpdate statusUpdate = new StatusUpdate(StatusUpdate.STATUS_CANCELLED);
			try {
				producer.send(JobMessageFactory.getUpdateStatusMessage(consumerRecord.key(), statusUpdate, space));
			} catch (JsonProcessingException jsonException) {
				error = String.format(
						"Error sending Cancelled Status from Job %s: %s. The Job was cancelled, but its status will not be updated in the Job Manager.",
						consumerRecord.key(), jsonException.getMessage());
				LOGGER.error(error, jsonException);
				pzLogger.log(error, Severity.ERROR);
			}
		} catch (Exception exception) {
			String error = String.format("Error Accessing Data under Job %s with Error: %s", consumerRecord.key(), exception.getMessage());
			LOGGER.error(error, exception, new AuditElement(consumerRecord.key(), "failedAccessData", ""));
			pzLogger.log(error, Severity.ERROR);

			try {
				// Send the failure message to the Job Manager.
				StatusUpdate statusUpdate = new StatusUpdate(StatusUpdate.STATUS_ERROR);
				statusUpdate.setResult(new ErrorResult("Could not Deploy Data", exception.getMessage()));
				producer.send(JobMessageFactory.getUpdateStatusMessage(consumerRecord.key(), statusUpdate, space));
			} catch (JsonProcessingException jsonException) {
				// If the Kafka message fails to send, at least log
				// something in the console.
				String errorJson = String.format(
						"Could not update Job Manager with failure event in Ingest Worker. Error creating message: %s",
						jsonException.getMessage());
				LOGGER.error(errorJson, jsonException);
				pzLogger.log(errorJson, Severity.ERROR, new AuditElement(consumerRecord.key(), "failedAccessData", errorJson));
			}
		} finally {
			if (callback != null) {
				callback.onComplete(consumerRecord.key());
			}
		}

		return new AsyncResult<>(accessJob);
	}

	private void processGeoServerType(Job job, AccessJob accessJob, Producer<String, String> producer, String key)
			throws JsonProcessingException, InvalidInputException, InterruptedException, GeoServerException, DataInspectException {<commit_msg>27218 Update Audit Log<commit_after>/**
f_ * Copyright 2016, RadiantBlue Technologies, Inc.
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 **/
package access.messaging;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Future;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.Producer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Component;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;

import access.database.DatabaseAccessor;
import access.deploy.Deployer;
import access.deploy.GroupDeployer;
import access.deploy.Leaser;
import exception.DataInspectException;
import exception.GeoServerException;
import exception.InvalidInputException;
import messaging.job.JobMessageFactory;
import messaging.job.WorkerCallback;
import model.data.DataResource;
import model.data.deployment.Deployment;
import model.data.deployment.DeploymentGroup;
import model.job.Job;
import model.job.result.type.DeploymentResult;
import model.job.result.type.ErrorResult;
import model.job.type.AccessJob;
import model.logger.AuditElement;
import model.logger.Severity;
import model.status.StatusUpdate;
import util.PiazzaLogger;

/**
 * Worker class that handles Access Jobs being passed in through the Kafka. Handles the Access Jobs by standing up
 * services, retrieving files, or other various forms of Access for data.
 * 
 * This component assumes that the data intended to be accessed is already ingested into the Piazza system; either by
 * the Ingest component or other components that are capable of inserting data into Piazza.
 * 
 * @author Patrick.Doody & Sonny.Saniev
 * 
 */
@Component
public class AccessWorker {
	@Autowired
	private Deployer deployer;
	@Autowired
	private GroupDeployer groupDeployer;
	@Autowired
	private DatabaseAccessor accessor;
	@Autowired
	private Leaser leaser;
	@Autowired
	private PiazzaLogger pzLogger;
	@Value("${SPACE}")
	private String space;

	private static final Logger LOGGER = LoggerFactory.getLogger(AccessWorker.class);

	/**
	 * Listens for Kafka Access messages for creating Deployments for Access of Resources
	 */
	@Async
	public Future<AccessJob> run(ConsumerRecord<String, String> consumerRecord, Producer<String, String> producer,
			WorkerCallback callback) {
		AccessJob accessJob = null;
		try {
			// Parse Job information from Kafka
			ObjectMapper mapper = new ObjectMapper();
			Job job = mapper.readValue(consumerRecord.value(), Job.class);
			accessJob = (AccessJob) job.getJobType();

			// Validate inputs for the Kafka Message
			if ((accessJob.getDataId() == null) || (accessJob.getDataId().isEmpty())) {
				throw new InvalidInputException(String.format("An invalid or empty Data Id was specified: %s", accessJob.getDataId()));
			}

			if ((accessJob.getDeploymentType() == null) || (accessJob.getDeploymentType().isEmpty())) {
				throw new InvalidInputException(
						String.format("An invalid or empty Deployment Type was specified: %s", accessJob.getDataId()));
			}

			// Logging
			pzLogger.log(
					String.format("Received Request to Access Data %s of Type %s under Job Id %s by user %s", accessJob.getDataId(),
							accessJob.getDeploymentType(), job.getJobId(), job.getCreatedBy()),
					Severity.INFORMATIONAL, new AuditElement(job.getJobId(), "requestAccessData", accessJob.getDataId()));

			if (Thread.interrupted()) {
				throw new InterruptedException();
			}

			processGeoServerType(job, accessJob, producer, consumerRecord.key());

		} catch (InterruptedException exception) {
			String error = String.format("Thread interrupt received for Job %s", consumerRecord.key());
			LOGGER.error(error, exception);
			pzLogger.log(error, Severity.INFORMATIONAL, new AuditElement(consumerRecord.key(), "accessJobTerminated", ""));
			StatusUpdate statusUpdate = new StatusUpdate(StatusUpdate.STATUS_CANCELLED);
			try {
				producer.send(JobMessageFactory.getUpdateStatusMessage(consumerRecord.key(), statusUpdate, space));
			} catch (JsonProcessingException jsonException) {
				error = String.format(
						"Error sending Cancelled Status from Job %s: %s. The Job was cancelled, but its status will not be updated in the Job Manager.",
						consumerRecord.key(), jsonException.getMessage());
				LOGGER.error(error, jsonException);
				pzLogger.log(error, Severity.ERROR, new AuditElement(consumerRecord.key(), "failedToSendCancelledStatus", ""));
			}
		} catch (Exception exception) {
			String error = String.format("Error Accessing Data under Job %s with Error: %s", consumerRecord.key(), exception.getMessage());
			LOGGER.error(error, exception, new AuditElement(consumerRecord.key(), "failedAccessData", ""));
			pzLogger.log(error, Severity.ERROR);

			try {
				// Send the failure message to the Job Manager.
				StatusUpdate statusUpdate = new StatusUpdate(StatusUpdate.STATUS_ERROR);
				statusUpdate.setResult(new ErrorResult("Could not Deploy Data", exception.getMessage()));
				producer.send(JobMessageFactory.getUpdateStatusMessage(consumerRecord.key(), statusUpdate, space));
			} catch (JsonProcessingException jsonException) {
				// If the Kafka message fails to send, at least log
				// something in the console.
				String errorJson = String.format(
						"Could not update Job Manager with failure event in Ingest Worker. Error creating message: %s",
						jsonException.getMessage());
				LOGGER.error(errorJson, jsonException);
				pzLogger.log(errorJson, Severity.ERROR, new AuditElement(consumerRecord.key(), "failedAccessData", errorJson));
			}
		} finally {
			if (callback != null) {
				callback.onComplete(consumerRecord.key());
			}
		}

		return new AsyncResult<>(accessJob);
	}

	private void processGeoServerType(Job job, AccessJob accessJob, Producer<String, String> producer, String key)
			throws JsonProcessingException, InvalidInputException, InterruptedException, GeoServerException, DataInspectException {
{'fraction_non_alphanumeric': 0.09958382877526754, 'fraction_numerical': 0.0018579072532699168, 'mean_word_length': 5.004908522980812, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 4}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5100609', 'n_tokens_mistral': 4047, 'n_tokens_neox': 3664, 'n_words': 1102}