<commit_before>/*
 * Copyright 2016 SimplifyOps, Inc. (http://simplifyops.com)
 *
 * 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 rundeck.services

import com.codahale.metrics.Counter
import com.dtolabs.rundeck.app.internal.logging.FSStreamingLogReader
import com.dtolabs.rundeck.app.internal.logging.FSStreamingLogWriter
import com.dtolabs.rundeck.app.internal.logging.RundeckLogFormat
import com.dtolabs.rundeck.app.internal.workflow.PeriodicFileChecker
import com.dtolabs.rundeck.core.logging.ExecutionFileStorage
import com.dtolabs.rundeck.core.logging.ExecutionFileStorageOptions
import com.dtolabs.rundeck.core.logging.ExecutionMultiFileStorage
import com.dtolabs.rundeck.core.logging.LogFileState
import com.dtolabs.rundeck.core.logging.ExecutionFileStorageException
import com.dtolabs.rundeck.core.logging.StreamingLogReader
import com.dtolabs.rundeck.core.logging.StreamingLogWriter
import com.dtolabs.rundeck.core.plugins.configuration.PropertyResolver
import com.dtolabs.rundeck.core.plugins.configuration.PropertyScope
import com.dtolabs.rundeck.plugins.logging.ExecutionFileStoragePlugin
import com.dtolabs.rundeck.server.plugins.services.ExecutionFileStoragePluginProviderService
import org.hibernate.sql.JoinType
import org.springframework.beans.factory.InitializingBean
import org.springframework.context.ApplicationContext
import org.springframework.context.ApplicationContextAware
import org.springframework.core.task.AsyncTaskExecutor
import org.springframework.scheduling.TaskScheduler
import rundeck.Execution
import rundeck.LogFileStorageRequest
import rundeck.services.events.ExecutionCompleteEvent
import rundeck.services.execution.ValueHolder
import rundeck.services.execution.ValueWatcher
import rundeck.services.logging.ExecutionFile
import rundeck.services.logging.ExecutionFileProducer
import rundeck.services.logging.ExecutionFileUtil
import rundeck.services.logging.ExecutionLogReader
import rundeck.services.logging.ExecutionLogState
import rundeck.services.logging.LogFileLoader
import rundeck.services.logging.MultiFileStorageRequestImpl

import java.util.concurrent.BlockingQueue
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.ConcurrentLinkedQueue
import java.util.concurrent.Executors
import java.util.concurrent.LinkedBlockingQueue
import java.util.concurrent.ScheduledExecutorService
import java.util.concurrent.TimeUnit
import java.nio.file.Files
import java.nio.file.StandardCopyOption

/**
 * Manage execution file storage retrieve and store requests.
 * "executorService" runs tasks within a hibernate session.
 * "logFileTaskExecutor" runs asynchronous tasks as well as two threads which process retrieve/storage queues
 * "scheduledExecutor" runs delayed tasks for retrying at a later time, OR runs periodic queue processing of resumed tasks
 *     this depends on whether using 'periodic' or 'delayed' strategy, default 'periodic'.
 * "retryIncompleteRequests" queue for resumed incomplete requests
 * "storageRequests" blocking queue for storage requests
 * "retrievalRequests" blocking queue for retrieval requests
 */
class LogFileStorageService implements InitializingBean,ApplicationContextAware{

    static transactional = false
    static final RundeckLogFormat rundeckLogFormat = new RundeckLogFormat()
    ExecutionFileStoragePluginProviderService executionFileStoragePluginProviderService
    PluginService pluginService
    def frameworkService
    AsyncTaskExecutor logFileTaskExecutor
    AsyncTaskExecutor logFileStorageTaskExecutor
    TaskScheduler logFileStorageTaskScheduler
    def executorService
    def grailsApplication
    def grailsLinkGenerator
    ApplicationContext applicationContext
    def metricService
    def configurationService
    def grailsEvents

    /**
     * Queue of log storage requests ids, for incomplet requests being resumed
     */
    private BlockingQueue<Long> retryIncompleteRequests = new LinkedBlockingQueue<>()
    /**
     * Request IDs that were attempted and then failed
     */
    private Set<Long> failedRequests = new HashSet<>()
    private Map<Long,List<String>> failures = new HashMap<>()
    /**
     * Queue of log storage requests, processed by {@link #runStorageRequest(java.util.Map)}
     */
    private BlockingQueue<Map> storageRequests = new LinkedBlockingQueue<Map>()
    /**
     * Queue of log retrieval requests, processed by {@link #runRetrievalRequest(java.util.Map)}
     */
    private BlockingQueue<Map> retrievalRequests = new LinkedBlockingQueue<Map>()
    /**
     * Currently running requests
     */
    private Queue<Map> running = new ConcurrentLinkedQueue<Map>()
    /**
     * Map of log retrieval actions
     */
    protected ConcurrentHashMap<String, Map> logFileRetrievalRequests = new ConcurrentHashMap<String, Map>()
    protected ConcurrentHashMap<String, Map> logFileRetrievalResults = new ConcurrentHashMap<String, Map>()
    @Override
    void afterPropertiesSet() throws Exception {
        def pluginName = getConfiguredPluginName()
        if(!pluginName){
            //System.err.println("LogFileStoragePlugin not configured, disabling...")
            return
        }
        logFileStorageTaskExecutor?.execute(new TaskRunner<Map>(storageRequests, { Map task ->
            storageQueueCounter?.dec()

            //run within same executor on another thread, may block until a thread is available
            logFileStorageTaskExecutor.execute {

                runStorageRequest(task)
            }
        }))
        logFileTaskExecutor?.execute( new TaskRunner<Map>(retrievalRequests,{ Map task ->
            runRetrievalRequest(task)
        }))
        if (getConfiguredResumeStrategy() == 'periodic') {
            long delay = getConfiguredStorageRetryDelay() * 1000
            logFileStorageTaskScheduler.scheduleAtFixedRate(this.&dequeueIncompleteLogStorage, new Date(System.currentTimeMillis() + delay), delay)
        }
    }<commit_msg>eval concurrency settings within afterPropertiesSet<commit_after>/*
 * Copyright 2016 SimplifyOps, Inc. (http://simplifyops.com)
 *
 * 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 rundeck.services

import com.codahale.metrics.Counter
import com.dtolabs.rundeck.app.internal.logging.FSStreamingLogReader
import com.dtolabs.rundeck.app.internal.logging.FSStreamingLogWriter
import com.dtolabs.rundeck.app.internal.logging.RundeckLogFormat
import com.dtolabs.rundeck.app.internal.workflow.PeriodicFileChecker
import com.dtolabs.rundeck.core.logging.ExecutionFileStorage
import com.dtolabs.rundeck.core.logging.ExecutionFileStorageOptions
import com.dtolabs.rundeck.core.logging.ExecutionMultiFileStorage
import com.dtolabs.rundeck.core.logging.LogFileState
import com.dtolabs.rundeck.core.logging.ExecutionFileStorageException
import com.dtolabs.rundeck.core.logging.StreamingLogReader
import com.dtolabs.rundeck.core.logging.StreamingLogWriter
import com.dtolabs.rundeck.core.plugins.configuration.PropertyResolver
import com.dtolabs.rundeck.core.plugins.configuration.PropertyScope
import com.dtolabs.rundeck.plugins.logging.ExecutionFileStoragePlugin
import com.dtolabs.rundeck.server.plugins.services.ExecutionFileStoragePluginProviderService
import org.hibernate.sql.JoinType
import org.springframework.beans.factory.InitializingBean
import org.springframework.context.ApplicationContext
import org.springframework.context.ApplicationContextAware
import org.springframework.core.task.AsyncTaskExecutor
import org.springframework.scheduling.TaskScheduler
import rundeck.Execution
import rundeck.LogFileStorageRequest
import rundeck.services.events.ExecutionCompleteEvent
import rundeck.services.execution.ValueHolder
import rundeck.services.execution.ValueWatcher
import rundeck.services.logging.ExecutionFile
import rundeck.services.logging.ExecutionFileProducer
import rundeck.services.logging.ExecutionFileUtil
import rundeck.services.logging.ExecutionLogReader
import rundeck.services.logging.ExecutionLogState
import rundeck.services.logging.LogFileLoader
import rundeck.services.logging.MultiFileStorageRequestImpl

import java.util.concurrent.BlockingQueue
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.ConcurrentLinkedQueue
import java.util.concurrent.Executors
import java.util.concurrent.LinkedBlockingQueue
import java.util.concurrent.ScheduledExecutorService
import java.util.concurrent.TimeUnit
import java.nio.file.Files
import java.nio.file.StandardCopyOption

/**
 * Manage execution file storage retrieve and store requests.
 * "executorService" runs tasks within a hibernate session.
 * "logFileTaskExecutor" runs asynchronous tasks as well as two threads which process retrieve/storage queues
 * "scheduledExecutor" runs delayed tasks for retrying at a later time, OR runs periodic queue processing of resumed tasks
 *     this depends on whether using 'periodic' or 'delayed' strategy, default 'periodic'.
 * "retryIncompleteRequests" queue for resumed incomplete requests
 * "storageRequests" blocking queue for storage requests
 * "retrievalRequests" blocking queue for retrieval requests
 */
class LogFileStorageService implements InitializingBean,ApplicationContextAware{

    static transactional = false
    static final RundeckLogFormat rundeckLogFormat = new RundeckLogFormat()
    ExecutionFileStoragePluginProviderService executionFileStoragePluginProviderService
    PluginService pluginService
    def frameworkService
    AsyncTaskExecutor logFileTaskExecutor
    AsyncTaskExecutor logFileStorageTaskExecutor
    TaskScheduler logFileStorageTaskScheduler
    def executorService
    def grailsApplication
    def grailsLinkGenerator
    ApplicationContext applicationContext
    def metricService
    def configurationService
    def grailsEvents

    /**
     * Queue of log storage requests ids, for incomplet requests being resumed
     */
    private BlockingQueue<Long> retryIncompleteRequests = new LinkedBlockingQueue<>()
    /**
     * Request IDs that were attempted and then failed
     */
    private Set<Long> failedRequests = new HashSet<>()
    private Map<Long,List<String>> failures = new HashMap<>()
    /**
     * Queue of log storage requests, processed by {@link #runStorageRequest(java.util.Map)}
     */
    private BlockingQueue<Map> storageRequests = new LinkedBlockingQueue<Map>()
    /**
     * Queue of log retrieval requests, processed by {@link #runRetrievalRequest(java.util.Map)}
     */
    private BlockingQueue<Map> retrievalRequests = new LinkedBlockingQueue<Map>()
    /**
     * Currently running requests
     */
    private Queue<Map> running = new ConcurrentLinkedQueue<Map>()
    /**
     * Map of log retrieval actions
     */
    protected ConcurrentHashMap<String, Map> logFileRetrievalRequests = new ConcurrentHashMap<String, Map>()
    protected ConcurrentHashMap<String, Map> logFileRetrievalResults = new ConcurrentHashMap<String, Map>()
    @Override
    void afterPropertiesSet() throws Exception {
        def pluginName = getConfiguredPluginName()
        if(!pluginName){
            //System.err.println("LogFileStoragePlugin not configured, disabling...")
            return
        }

        logFileStorageTaskExecutor.concurrencyLimit = 1 + configurationService.getInteger('execution.logs.fileStorage.storageTasks.concurrencyLimit', 5)
        logFileTaskExecutor.concurrencyLimit = 1 + configurationService.getInteger('execution.logs.fileStorage.retrievalTasks.concurrencyLimit', 5)

        log.debug("logFileStorageTaskExecutor concurrency: ${logFileStorageTaskExecutor.concurrencyLimit}")
        log.debug("logFileTaskExecutor concurrency: ${logFileTaskExecutor.concurrencyLimit}")
        logFileStorageTaskExecutor?.execute(new TaskRunner<Map>(storageRequests, { Map task ->
            storageQueueCounter?.dec()

            //run within same executor on another thread, may block until a thread is available
            logFileStorageTaskExecutor.execute {

                runStorageRequest(task)
            }
        }))
        logFileTaskExecutor?.execute( new TaskRunner<Map>(retrievalRequests,{ Map task ->
            runRetrievalRequest(task)
        }))
        if (getConfiguredResumeStrategy() == 'periodic') {
            long delay = getConfiguredStorageRetryDelay() * 1000
            logFileStorageTaskScheduler.scheduleAtFixedRate(this.&dequeueIncompleteLogStorage, new Date(System.currentTimeMillis() + delay), delay)
        }
    }
{'fraction_non_alphanumeric': 0.07325952893974293, 'fraction_numerical': 0.0020803923025484805, 'mean_word_length': 4.710649130250318, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 45, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, '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': '5454247', 'n_tokens_mistral': 3483, 'n_tokens_neox': 3361, 'n_words': 874}