/**
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.  The ASF licenses this file to you 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 org.apache.storm.metricstore.rocksdb;

import com.codahale.metrics.Meter;
import java.io.File;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.storm.DaemonConfig;
import org.apache.storm.metric.StormMetricsRegistry;
import org.apache.storm.metricstore.AggLevel;
import org.apache.storm.metricstore.FilterOptions;
import org.apache.storm.metricstore.Metric;
import org.apache.storm.metricstore.MetricException;
import org.apache.storm.metricstore.MetricStore;
import org.apache.storm.utils.ConfigUtils;
import org.apache.storm.utils.ObjectReader;
import org.rocksdb.BlockBasedTableConfig;
import org.rocksdb.IndexType;
import org.rocksdb.Options;
import org.rocksdb.ReadOptions;
import org.rocksdb.RocksDB;
import org.rocksdb.RocksDBException;
import org.rocksdb.RocksIterator;
import org.rocksdb.WriteBatch;
import org.rocksdb.WriteOptions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


public class RocksDbStore implements MetricStore, AutoCloseable {
    static final int INVALID_METADATA_STRING_ID = 0;
    private static final Logger LOG = LoggerFactory.getLogger(RocksDbStore.class);
    private static final int MAX_QUEUE_CAPACITY = 4000;
    RocksDB db;
    private ReadOnlyStringMetadataCache readOnlyStringMetadataCache = null;
    private BlockingQueue queue = new LinkedBlockingQueue(MAX_QUEUE_CAPACITY);
    private RocksDbMetricsWriter metricsWriter = null;
    private MetricsCleaner metricsCleaner = null;
    private Meter failureMeter = null;

    /**
     * Create metric store instance using the configurations provided via the config map.
     *
     * @param config Storm config map
     * @param metricsRegistry The Nimbus daemon metrics registry
     * @throws MetricException on preparation error
     */
    @Override
    public void prepare(Map<String, Object> config, StormMetricsRegistry metricsRegistry) throws MetricException {
        validateConfig(config);

        this.failureMeter = metricsRegistry.registerMeter("RocksDB:metric-failures");

        RocksDB.loadLibrary();
        boolean createIfMissing = ObjectReader.getBoolean(config.get(DaemonConfig.STORM_ROCKSDB_CREATE_IF_MISSING), false);

        try (Options options = new Options().setCreateIfMissing(createIfMissing)) {
            // use the hash index for prefix searches
            BlockBasedTableConfig tfc = new BlockBasedTableConfig();
            tfc.setIndexType(IndexType.kHashSearch);
            options.setTableFormatConfig(tfc);
            options.useCappedPrefixExtractor(RocksDbKey.KEY_SIZE);

            String path = getRocksDbAbsoluteDir(config);
            LOG.info("Opening RocksDB from {}, {}={}", path, DaemonConfig.STORM_ROCKSDB_CREATE_IF_MISSING, createIfMissing);
            db = RocksDB.open(options, path);
        } catch (RocksDBException e) {
            String message = "Error opening RockDB database";
            LOG.error(message, e);
            throw new MetricException(message, e);
        }

        // create thread to delete old metrics and metadata
        Integer retentionHours = Integer.parseInt(config.get(DaemonConfig.STORM_ROCKSDB_METRIC_RETENTION_HOURS).toString());
        Integer deletionPeriod = 0;
        if (config.containsKey(DaemonConfig.STORM_ROCKSDB_METRIC_DELETION_PERIOD_HOURS)) {
            deletionPeriod = Integer.parseInt(config.get(DaemonConfig.STORM_ROCKSDB_METRIC_DELETION_PERIOD_HOURS).toString());
        }
        metricsCleaner = new MetricsCleaner(this, retentionHours, deletionPeriod, failureMeter, metricsRegistry);

        // create thread to process insertion of all metrics
        metricsWriter = new RocksDbMetricsWriter(this, this.queue, this.failureMeter);

        int cacheCapacity = Integer.parseInt(config.get(DaemonConfig.STORM_ROCKSDB_METADATA_STRING_CACHE_CAPACITY).toString());
        StringMetadataCache.init(metricsWriter, cacheCapacity);
        readOnlyStringMetadataCache = StringMetadataCache.getReadOnlyStringMetadataCache();
        metricsWriter.init(); // init the writer once the cache is setup

        // start threads after metadata cache created
        Thread thread = new Thread(metricsCleaner, "RocksDbMetricsCleaner");
        thread.setDaemon(true);
        thread.start();

        thread = new Thread(metricsWriter, "RocksDbMetricsWriter");
        thread.setDaemon(true);
        thread.start();
    }

    /**
     * Implements configuration validation of Metrics Store, validates storm configuration for Metrics Store.
     *
     * @param config Storm config to specify which store type, location of store and creation policy
     * @throws MetricException if there is a missing required configuration or if the store does not exist but
     *                         the config specifies not to create the store
     */
    private void validateConfig(Map<String, Object> config) throws MetricException {
        if (!(config.containsKey(DaemonConfig.STORM_ROCKSDB_LOCATION))) {
            throw new MetricException("Not a vaild RocksDB configuration - Missing store location " + DaemonConfig.STORM_ROCKSDB_LOCATION);
        }

        if (!(config.containsKey(DaemonConfig.STORM_ROCKSDB_CREATE_IF_MISSING))) {
            throw new MetricException("Not a vaild RocksDB configuration - Does not specify creation policy "
                                      + DaemonConfig.STORM_ROCKSDB_CREATE_IF_MISSING);
        }

        // validate path defined
        String storePath = getRocksDbAbsoluteDir(config);

        boolean createIfMissing = ObjectReader.getBoolean(config.get(DaemonConfig.STORM_ROCKSDB_CREATE_IF_MISSING), false);
        if (!createIfMissing) {
            if (!(new File(storePath).exists())) {
                throw new MetricException("Configuration specifies not to create a store but no store currently exists at " + storePath);
            }
        }

        if (!(config.containsKey(DaemonConfig.STORM_ROCKSDB_METADATA_STRING_CACHE_CAPACITY))) {
            throw new MetricException("Not a valid RocksDB configuration - Missing metadata string cache size "
                                      + DaemonConfig.STORM_ROCKSDB_METADATA_STRING_CACHE_CAPACITY);
        }

        if (!config.containsKey(DaemonConfig.STORM_ROCKSDB_METRIC_RETENTION_HOURS)) {
            throw new MetricException("Not a valid RocksDB configuration - Missing metric retention "
                                      + DaemonConfig.STORM_ROCKSDB_METRIC_RETENTION_HOURS);
        }
    }

    private String getRocksDbAbsoluteDir(Map<String, Object> conf) throws MetricException {
        String storePath = (String) conf.get(DaemonConfig.STORM_ROCKSDB_LOCATION);
        if (storePath == null) {
            throw new MetricException("Not a vaild RocksDB configuration - Missing store location " + DaemonConfig.STORM_ROCKSDB_LOCATION);
        } else {
            if (new File(storePath).isAbsolute()) {
                return storePath;
            } else {
                String stormHome = System.getProperty(ConfigUtils.STORM_HOME);
                if (stormHome == null) {
                    throw new MetricException(ConfigUtils.STORM_HOME + " not set");
                }
                return (stormHome + File.separator + storePath);
            }
        }
    }

    /**
     * Stores metrics in the store.
     *
     * @param metric  Metric to store
     * @throws MetricException  if database write fails
     */
    @Override
    public void insert(Metric metric) throws MetricException {
        try {
            // don't bother blocking on a full queue, just drop metrics in case we can't keep up
            if (queue.remainingCapacity() <= 0) {
                LOG.info("Metrics q full, dropping metric");
                return;
            }
            queue.put(metric);
        } catch (Exception e) {
            String message = "Failed to insert metric";
            LOG.error(message, e);
            if (this.failureMeter != null) {
                this.failureMeter.mark();
            }
            throw new MetricException(message, e);
        }
    }

    /**
     * Fill out the numeric values for a metric.
     *
     * @param metric  Metric to populate
     * @return true if the metric was populated, false otherwise
     * @throws MetricException  if read from database fails
     */
    @Override
    public boolean populateValue(Metric metric) throws MetricException {
        Map<String, Integer> localLookupCache = new HashMap<>(6);

        int topologyId = lookupMetadataString(KeyType.TOPOLOGY_STRING, metric.getTopologyId(), localLookupCache);
        if (INVALID_METADATA_STRING_ID == topologyId) {
            return false;
        }
        int metricId = lookupMetadataString(KeyType.METRIC_STRING, metric.getMetricName(), localLookupCache);
        if (INVALID_METADATA_STRING_ID == metricId) {
            return false;
        }
        int componentId = lookupMetadataString(KeyType.COMPONENT_STRING, metric.getComponentId(), localLookupCache);
        if (INVALID_METADATA_STRING_ID == componentId) {
            return false;
        }
        int executorId = lookupMetadataString(KeyType.EXEC_ID_STRING, metric.getExecutorId(), localLookupCache);
        if (INVALID_METADATA_STRING_ID == executorId) {
            return false;
        }
        int hostId = lookupMetadataString(KeyType.HOST_STRING, metric.getHostname(), localLookupCache);
        if (INVALID_METADATA_STRING_ID == hostId) {
            return false;
        }
        int streamId = lookupMetadataString(KeyType.STREAM_ID_STRING, metric.getStreamId(), localLookupCache);
        if (INVALID_METADATA_STRING_ID == streamId) {
            return false;
        }

        RocksDbKey key = RocksDbKey.createMetricKey(metric.getAggLevel(), topologyId, metric.getTimestamp(), metricId,
                                                    componentId, executorId, hostId, metric.getPort(), streamId);

        return populateFromKey(key, metric);
    }

    // populate metric values using the provided key
    boolean populateFromKey(RocksDbKey key, Metric metric) throws MetricException {
        try {
            byte[] value = db.get(key.getRaw());
            if (value == null) {
                return false;
            }
            RocksDbValue rdbValue = new RocksDbValue(value);
            rdbValue.populateMetric(metric);
        } catch (Exception e) {
            String message = "Failed to populate metric";
            LOG.error(message, e);
            if (this.failureMeter != null) {
                this.failureMeter.mark();
            }
            throw new MetricException(message, e);
        }
        return true;
    }

    // attempts to lookup the unique Id for a string that may not exist yet.  Returns INVALID_METADATA_STRING_ID
    // if it does not exist.
    private int lookupMetadataString(KeyType type, String s, Map<String, Integer> lookupCache) throws MetricException {
        if (s == null) {
            if (this.failureMeter != null) {
                this.failureMeter.mark();
            }
            throw new MetricException("No string for metric metadata string type " + type);
        }

        // attempt to find it in the string cache, this will update the LRU
        StringMetadata stringMetadata = readOnlyStringMetadataCache.get(s);
        if (stringMetadata != null) {
            return stringMetadata.getStringId();
        }

        // attempt to find it in callers cache
        Integer id = lookupCache.get(s);
        if (id != null) {
            return id;
        }

        // attempt to find the string in the database
        try {
            stringMetadata = rocksDbGetStringMetadata(type, s);
        } catch (RocksDBException e) {
            throw new MetricException("Error reading metric data", e);
        }

        if (stringMetadata != null) {
            id = stringMetadata.getStringId();

            // add to the callers cache.  We can't add it to the stringMetadataCache, since that could cause an eviction
            // database write, which we want to only occur from the inserting DB thread.
            lookupCache.put(s, id);

            return id;
        }

        // string does not exist
        return INVALID_METADATA_STRING_ID;
    }

    // scans the database to look for a metadata string and returns the metadata info
    StringMetadata rocksDbGetStringMetadata(KeyType type, String s) throws RocksDBException {
        RocksDbKey firstKey = RocksDbKey.getInitialKey(type);
        RocksDbKey lastKey = RocksDbKey.getLastKey(type);
        final AtomicReference<StringMetadata> reference = new AtomicReference<>();
        scanRange(firstKey, lastKey, (key, value) -> {
            if (s.equals(value.getMetdataString())) {
                reference.set(value.getStringMetadata(key));
                return false;
            } else {
                return true;  // haven't found string, keep searching
            }
        });
        return reference.get();
    }

    // scans from key start to the key before end, calling back until callback indicates not to process further
    void scanRange(RocksDbKey start, RocksDbKey end, RocksDbScanCallback fn) throws RocksDBException {
        try (ReadOptions ro = new ReadOptions()) {
            ro.setTotalOrderSeek(true);
            try (RocksIterator iterator = db.newIterator(ro)) {
                for (iterator.seek(start.getRaw()); iterator.isValid(); iterator.next()) {
                    RocksDbKey key = new RocksDbKey(iterator.key());
                    if (key.compareTo(end) >= 0) { // past limit, quit
                        return;
                    }

                    RocksDbValue val = new RocksDbValue(iterator.value());
                    if (!fn.cb(key, val)) {
                        // if cb returns false, we are done with this section of rows
                        return;
                    }
                }
            }
        }
    }

    /**
     * Shutdown the store.
     */
    @Override
    public void close() {
        metricsWriter.close();
        metricsCleaner.close();
    }

    /**
     *  Scans all metrics in the store and returns the ones matching the specified filtering options.
     *  Callback returns Metric class results.
     *
     * @param filter   options to filter by
     * @param scanCallback  callback for each Metric found
     * @throws MetricException  on error
     */
    @Override
    public void scan(FilterOptions filter, ScanCallback scanCallback) throws MetricException {
        scanInternal(filter, scanCallback, null);
    }

    /**
     *  Scans all metrics in the store and returns the ones matching the specified filtering options.
     *  Callback returns raw key/value data.
     *
     * @param filter   options to filter by
     * @param rawCallback  callback for each Metric found
     * @throws MetricException  on error
     */
    private void scanRaw(FilterOptions filter, RocksDbScanCallback rawCallback) throws MetricException {
        scanInternal(filter, null, rawCallback);
    }

    // perform a scan given filter options, and return results in either Metric or raw data.
    private void scanInternal(FilterOptions filter, ScanCallback scanCallback, RocksDbScanCallback rawCallback) throws MetricException {

        Map<String, Integer> stringToIdCache = new HashMap<>();
        Map<Integer, String> idToStringCache = new HashMap<>();

        int startTopologyId = 0;
        int endTopologyId = 0xFFFFFFFF;
        String filterTopologyId = filter.getTopologyId();
        if (filterTopologyId != null) {
            int topologyId = lookupMetadataString(KeyType.TOPOLOGY_STRING, filterTopologyId, stringToIdCache);
            if (INVALID_METADATA_STRING_ID == topologyId) {
                return;  // string does not exist in database
            }
            startTopologyId = topologyId;
            endTopologyId = topologyId;
        }

        long startTime = filter.getStartTime();
        long endTime = filter.getEndTime();

        int startMetricId = 0;
        int endMetricId = 0xFFFFFFFF;
        String filterMetricName = filter.getMetricName();
        if (filterMetricName != null) {
            int metricId = lookupMetadataString(KeyType.METRIC_STRING, filterMetricName, stringToIdCache);
            if (INVALID_METADATA_STRING_ID == metricId) {
                return;  // string does not exist in database
            }
            startMetricId = metricId;
            endMetricId = metricId;
        }

        int startComponentId = 0;
        int endComponentId = 0xFFFFFFFF;
        String filterComponentId = filter.getComponentId();
        if (filterComponentId != null) {
            int componentId = lookupMetadataString(KeyType.COMPONENT_STRING, filterComponentId, stringToIdCache);
            if (INVALID_METADATA_STRING_ID == componentId) {
                return;  // string does not exist in database
            }
            startComponentId = componentId;
            endComponentId = componentId;
        }

        int startExecutorId = 0;
        int endExecutorId = 0xFFFFFFFF;
        String filterExecutorName = filter.getExecutorId();
        if (filterExecutorName != null) {
            int executorId = lookupMetadataString(KeyType.EXEC_ID_STRING, filterExecutorName, stringToIdCache);
            if (INVALID_METADATA_STRING_ID == executorId) {
                return;  // string does not exist in database
            }
            startExecutorId = executorId;
            endExecutorId = executorId;
        }

        int startHostId = 0;
        int endHostId = 0xFFFFFFFF;
        String filterHostId = filter.getHostId();
        if (filterHostId != null) {
            int hostId = lookupMetadataString(KeyType.HOST_STRING, filterHostId, stringToIdCache);
            if (INVALID_METADATA_STRING_ID == hostId) {
                return;  // string does not exist in database
            }
            startHostId = hostId;
            endHostId = hostId;
        }

        int startPort = 0;
        int endPort = 0xFFFFFFFF;
        Integer filterPort = filter.getPort();
        if (filterPort != null) {
            startPort = filterPort;
            endPort = filterPort;
        }

        int startStreamId = 0;
        int endStreamId = 0xFFFFFFFF;
        String filterStreamId = filter.getStreamId();
        if (filterStreamId != null) {
            int streamId = lookupMetadataString(KeyType.HOST_STRING, filterStreamId, stringToIdCache);
            if (INVALID_METADATA_STRING_ID == streamId) {
                return;  // string does not exist in database
            }
            startStreamId = streamId;
            endStreamId = streamId;
        }

        try (ReadOptions ro = new ReadOptions()) {
            ro.setTotalOrderSeek(true);

            for (AggLevel aggLevel : filter.getAggLevels()) {

                RocksDbKey startKey = RocksDbKey.createMetricKey(aggLevel, startTopologyId, startTime, startMetricId,
                        startComponentId, startExecutorId, startHostId, startPort, startStreamId);
                RocksDbKey endKey = RocksDbKey.createMetricKey(aggLevel, endTopologyId, endTime, endMetricId,
                        endComponentId, endExecutorId, endHostId, endPort, endStreamId);

                try (RocksIterator iterator = db.newIterator(ro)) {
                    for (iterator.seek(startKey.getRaw()); iterator.isValid(); iterator.next()) {
                        RocksDbKey key = new RocksDbKey(iterator.key());

                        if (key.compareTo(endKey) > 0) { // past limit, quit
                            break;
                        }

                        if (startTopologyId != 0 && key.getTopologyId() != startTopologyId) {
                            continue;
                        }

                        long timestamp = key.getTimestamp();
                        if (timestamp < startTime || timestamp > endTime) {
                            continue;
                        }

                        if (startMetricId != 0 && key.getMetricId() != startMetricId) {
                            continue;
                        }

                        if (startComponentId != 0 && key.getComponentId() != startComponentId) {
                            continue;
                        }

                        if (startExecutorId != 0 && key.getExecutorId() != startExecutorId) {
                            continue;
                        }

                        if (startHostId != 0 && key.getHostnameId() != startHostId) {
                            continue;
                        }

                        if (startPort != 0 && key.getPort() != startPort) {
                            continue;
                        }

                        if (startStreamId != 0 && key.getStreamId() != startStreamId) {
                            continue;
                        }

                        RocksDbValue val = new RocksDbValue(iterator.value());

                        if (scanCallback != null) {
                            try {
                                // populate a metric
                                String metricName = metadataIdToString(KeyType.METRIC_STRING, key.getMetricId(), idToStringCache);
                                String topologyId = metadataIdToString(KeyType.TOPOLOGY_STRING, key.getTopologyId(), idToStringCache);
                                String componentId = metadataIdToString(KeyType.COMPONENT_STRING, key.getComponentId(), idToStringCache);
                                String executorId = metadataIdToString(KeyType.EXEC_ID_STRING, key.getExecutorId(), idToStringCache);
                                String hostname = metadataIdToString(KeyType.HOST_STRING, key.getHostnameId(), idToStringCache);
                                String streamId = metadataIdToString(KeyType.STREAM_ID_STRING, key.getStreamId(), idToStringCache);

                                Metric metric = new Metric(metricName, timestamp, topologyId, 0.0, componentId, executorId, hostname,
                                        streamId, key.getPort(), aggLevel);

                                val.populateMetric(metric);

                                // callback to caller
                                scanCallback.cb(metric);
                            } catch (MetricException e) {
                                LOG.warn("Failed to report found metric: {}", e.getMessage());
                            }
                        } else {
                            try {
                                if (!rawCallback.cb(key, val)) {
                                    return;
                                }
                            } catch (RocksDBException e) {
                                throw new MetricException("Error reading metrics data", e);
                            }
                        }
                    }
                }
            }
        }
    }

    // Finds the metadata string that matches the string Id and type provided.  The string should exist, as it is
    // referenced from a metric.
    private String metadataIdToString(KeyType type, int id, Map<Integer, String> lookupCache) throws MetricException {
        String s = readOnlyStringMetadataCache.getMetadataString(id);
        if (s != null) {
            return s;
        }
        s = lookupCache.get(id);
        if (s != null) {
            return s;
        }
        // get from DB and add to lookup cache
        RocksDbKey key = new RocksDbKey(type, id);
        try {
            byte[] value = db.get(key.getRaw());
            if (value == null) {
                throw new MetricException("Failed to find metadata string for id " + id + " of type " + type);
            }
            RocksDbValue rdbValue = new RocksDbValue(value);
            s = rdbValue.getMetdataString();
            lookupCache.put(id, s);
            return s;
        } catch (RocksDBException e) {
            if (this.failureMeter != null) {
                this.failureMeter.mark();
            }
            throw new MetricException("Failed to get from RocksDb", e);
        }
    }

    // deletes metrics matching the filter options
    void deleteMetrics(FilterOptions filter) throws MetricException {
        try (WriteBatch writeBatch = new WriteBatch();
             WriteOptions writeOps = new WriteOptions()) {

            scanRaw(filter, (RocksDbKey key, RocksDbValue value) -> {
                writeBatch.delete(key.getRaw());
                return true;
            });

            if (writeBatch.count() > 0) {
                LOG.info("Deleting {} metrics", writeBatch.count());
                try {
                    db.write(writeOps, writeBatch);
                } catch (Exception e) {
                    String message = "Failed delete metrics";
                    LOG.error(message, e);
                    if (this.failureMeter != null) {
                        this.failureMeter.mark();
                    }
                    throw new MetricException(message, e);
                }
            }
        }
    }

    // deletes metadata strings before the provided timestamp
    void deleteMetadataBefore(long firstValidTimestamp) throws MetricException {
        if (firstValidTimestamp < 1L) {
            if (this.failureMeter != null) {
                this.failureMeter.mark();
            }
            throw new MetricException("Invalid timestamp for deleting metadata: " + firstValidTimestamp);
        }

        try (WriteBatch writeBatch = new WriteBatch();
             WriteOptions writeOps = new WriteOptions()) {

            // search all metadata strings
            RocksDbKey topologyMetadataPrefix = RocksDbKey.getPrefix(KeyType.METADATA_STRING_START);
            RocksDbKey lastPrefix = RocksDbKey.getPrefix(KeyType.METADATA_STRING_END);
            try {
                scanRange(topologyMetadataPrefix, lastPrefix, (key, value) -> {
                    // we'll assume the metadata was recently used if still in the cache.
                    if (!readOnlyStringMetadataCache.contains(key.getMetadataStringId())) {
                        if (value.getLastTimestamp() < firstValidTimestamp) {
                            writeBatch.delete(key.getRaw());
                        }
                    }
                    return true;
                });
            } catch (RocksDBException e) {
                throw new MetricException("Error reading metric data", e);
            }

            if (writeBatch.count() > 0) {
                LOG.info("Deleting {} metadata strings", writeBatch.count());
                try {
                    db.write(writeOps, writeBatch);
                } catch (Exception e) {
                    String message = "Failed delete metadata strings";
                    LOG.error(message, e);
                    if (this.failureMeter != null) {
                        this.failureMeter.mark();
                    }
                    throw new MetricException(message, e);
                }
            }
        }
    }

    interface RocksDbScanCallback {
        boolean cb(RocksDbKey key, RocksDbValue val) throws RocksDBException;  // return false to stop scan
    }
}