package org.apache.blur.thrift;

/**
 * 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.
 */
import static org.apache.blur.utils.BlurConstants.BLUR_CLUSTER;
import static org.apache.blur.utils.BlurConstants.BLUR_CLUSTER_NAME;
import static org.apache.blur.utils.BlurConstants.BLUR_COMMAND_LIB_PATH;
import static org.apache.blur.utils.BlurConstants.BLUR_CONTROLLER_REMOTE_FETCH_COUNT;
import static org.apache.blur.utils.BlurConstants.BLUR_GC_BACK_PRESSURE_HEAP_RATIO;
import static org.apache.blur.utils.BlurConstants.BLUR_GUI_SHARD_PORT;
import static org.apache.blur.utils.BlurConstants.BLUR_HTTP_STATUS_RUNNING_PORT;
import static org.apache.blur.utils.BlurConstants.BLUR_INDEXMANAGER_FACET_THREAD_COUNT;
import static org.apache.blur.utils.BlurConstants.BLUR_INDEXMANAGER_MUTATE_THREAD_COUNT;
import static org.apache.blur.utils.BlurConstants.BLUR_INDEXMANAGER_SEARCH_THREAD_COUNT;
import static org.apache.blur.utils.BlurConstants.BLUR_MAX_CLAUSE_COUNT;
import static org.apache.blur.utils.BlurConstants.BLUR_MAX_HEAP_PER_ROW_FETCH;
import static org.apache.blur.utils.BlurConstants.BLUR_MAX_RECORDS_PER_ROW_FETCH_REQUEST;
import static org.apache.blur.utils.BlurConstants.BLUR_NODENAME;
import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_BIND_ADDRESS;
import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_BIND_PORT;
import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_BLOCK_CACHE_TOTAL_SIZE;
import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_BLOCK_CACHE_VERSION;
import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_COMMAND_DRIVER_THREADS;
import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_COMMAND_WORKER_THREADS;
import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_DEEP_PAGING_CACHE_SIZE;
import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_FETCHCOUNT;
import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_FILTER_CACHE_CLASS;
import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_HOSTNAME;
import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_INDEX_MAX_IDLE_TIME;
import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_INTERNAL_SEARCH_THREAD_COUNT;
import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_MERGE_THREAD_COUNT;
import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_OPENER_THREAD_COUNT;
import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_REQUEST_CACHE_SIZE;
import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_SAFEMODEDELAY;
import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_SERVER_MINIMUM_BEFORE_SAFEMODE_EXIT;
import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_SERVER_THRIFT_THREAD_COUNT;
import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_SMALL_MERGE_THRESHOLD;
import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_THRIFT_ACCEPT_QUEUE_SIZE_PER_THREAD;
import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_THRIFT_MAX_READ_BUFFER_BYTES;
import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_THRIFT_SELECTOR_THREADS;
import static org.apache.blur.utils.BlurConstants.BLUR_STREAM_SERVER_RUNNING_PORT;
import static org.apache.blur.utils.BlurConstants.BLUR_STREAM_SERVER_THREADS;
import static org.apache.blur.utils.BlurConstants.BLUR_THRIFT_DEFAULT_MAX_FRAME_SIZE;
import static org.apache.blur.utils.BlurConstants.BLUR_THRIFT_MAX_FRAME_SIZE;
import static org.apache.blur.utils.BlurUtil.quietClose;

import java.io.Closeable;
import java.io.File;
import java.lang.reflect.Constructor;
import java.util.List;
import java.util.Map.Entry;
import java.util.Set;
import java.util.Timer;
import java.util.concurrent.TimeUnit;

import org.apache.blur.BlurConfiguration;
import org.apache.blur.command.ShardCommandManager;
import org.apache.blur.command.stream.StreamProcessor;
import org.apache.blur.command.stream.StreamServer;
import org.apache.blur.concurrent.SimpleUncaughtExceptionHandler;
import org.apache.blur.concurrent.ThreadWatcher;
import org.apache.blur.gui.HttpJettyServer;
import org.apache.blur.log.Log;
import org.apache.blur.log.LogFactory;
import org.apache.blur.lucene.search.DeepPagingCache;
import org.apache.blur.manager.BlurFilterCache;
import org.apache.blur.manager.BlurQueryChecker;
import org.apache.blur.manager.DefaultBlurFilterCache;
import org.apache.blur.manager.IndexManager;
import org.apache.blur.manager.clusterstatus.ClusterStatus.Action;
import org.apache.blur.manager.clusterstatus.ZookeeperClusterStatus;
import org.apache.blur.manager.indexserver.BlurServerShutDown;
import org.apache.blur.manager.indexserver.BlurServerShutDown.BlurShutdown;
import org.apache.blur.manager.indexserver.DistributedIndexServer;
import org.apache.blur.manager.indexserver.DistributedLayoutFactory;
import org.apache.blur.manager.indexserver.DistributedLayoutFactoryImpl;
import org.apache.blur.manager.status.QueryStatusManager;
import org.apache.blur.memory.MemoryAllocationWatcher;
import org.apache.blur.memory.Watcher;
import org.apache.blur.metrics.JSONReporter;
import org.apache.blur.metrics.ReporterSetup;
import org.apache.blur.server.ServerSecurityFilter;
import org.apache.blur.server.ServerSecurityFilterFactory;
import org.apache.blur.server.ServerSecurityUtil;
import org.apache.blur.server.ShardServerEventHandler;
import org.apache.blur.server.TableContext;
import org.apache.blur.server.cache.ThriftCache;
import org.apache.blur.server.cache.ThriftCacheServer;
import org.apache.blur.store.BlockCacheDirectoryFactory;
import org.apache.blur.store.BlockCacheDirectoryFactoryV1;
import org.apache.blur.store.BlockCacheDirectoryFactoryV2;
import org.apache.blur.store.buffer.BufferStore;
import org.apache.blur.store.hdfs.SequentialReadControl;
import org.apache.blur.thirdparty.thrift_0_9_0.protocol.TJSONProtocol;
import org.apache.blur.thirdparty.thrift_0_9_0.server.TServlet;
import org.apache.blur.thirdparty.thrift_0_9_0.transport.TServerTransport;
import org.apache.blur.thrift.generated.Blur;
import org.apache.blur.thrift.generated.Blur.Iface;
import org.apache.blur.trace.Trace;
import org.apache.blur.trace.TraceStorage;
import org.apache.blur.utils.BlurUtil;
import org.apache.blur.utils.GCWatcher;
import org.apache.blur.utils.MemoryReporter;
import org.apache.hadoop.conf.Configuration;
import org.apache.lucene.search.BooleanQuery;
import org.apache.zookeeper.ZooKeeper;
import org.eclipse.jetty.servlet.ServletHolder;
import org.eclipse.jetty.webapp.WebAppContext;

import sun.misc.VM;

public class ThriftBlurShardServer extends ThriftServer {

  private static final Log LOG = LogFactory.getLog(ThriftBlurShardServer.class);
  private static final boolean enableJsonReporter = false;
  private static final long _64MB = 64 * 1024 * 1024;

  public static void main(String[] args) throws Exception {
    try {
      int serverIndex = getServerIndex(args);
      LOG.info("Setting up Shard Server");
      Thread.setDefaultUncaughtExceptionHandler(new SimpleUncaughtExceptionHandler());
      BlurConfiguration configuration = new BlurConfiguration();
      printUlimits();
      ReporterSetup.setupReporters(configuration);
      MemoryReporter.enable();
      setupJvmMetrics();
      double ratio = configuration.getDouble(BLUR_GC_BACK_PRESSURE_HEAP_RATIO, 0.75);
      GCWatcher.init(ratio);
      ThriftServer server = createServer(serverIndex, configuration);
      server.start();
    } catch (Throwable t) {
      t.printStackTrace();
      System.exit(1);
    }
  }

  public static ThriftServer createServer(int serverIndex, BlurConfiguration configuration) throws Exception {
    Configuration config = BlurUtil.newHadoopConfiguration(configuration);
    TableContext.setSystemBlurConfiguration(configuration);
    TableContext.setSystemConfiguration(config);

    String bindAddress = configuration.get(BLUR_SHARD_BIND_ADDRESS);
    int configBindPort = configuration.getInt(BLUR_SHARD_BIND_PORT, -1);
    int instanceBindPort = configBindPort + serverIndex;
    if (configBindPort == 0) {
      instanceBindPort = 0;
    }
    TServerTransport serverTransport = ThriftServer.getTServerTransport(bindAddress, instanceBindPort, configuration);
    instanceBindPort = ThriftServer.getBindingPort(serverTransport);

    Set<Entry<String, String>> set = configuration.getProperties().entrySet();
    for (Entry<String, String> e : set) {
      String key = e.getKey();
      String value = e.getValue();
      if (value == null || value.isEmpty()) {
        continue;
      }
      if (key.startsWith("blur.shard.buffercache.")) {
        int index = key.lastIndexOf('.');
        int bufferSize = Integer.parseInt(key.substring(index + 1));
        long amount = Long.parseLong(e.getValue());
        BufferStore.initNewBuffer(bufferSize, amount);
      }
    }

    final BlockCacheDirectoryFactory blockCacheDirectoryFactory;
    // Alternate BlockCacheDirectoryFactory support currently disabled in 0.2.0,
    // look for it in 0.2.1
    String blockCacheVersion = configuration.get(BLUR_SHARD_BLOCK_CACHE_VERSION, "v2");
    long totalNumberOfBytes = configuration.getLong(BLUR_SHARD_BLOCK_CACHE_TOTAL_SIZE, VM.maxDirectMemory() - _64MB);
    if (blockCacheVersion.equals("v1")) {
      blockCacheDirectoryFactory = new BlockCacheDirectoryFactoryV1(configuration, totalNumberOfBytes);
    } else if (blockCacheVersion.equals("v2")) {
      blockCacheDirectoryFactory = new BlockCacheDirectoryFactoryV2(configuration, totalNumberOfBytes);
    } else {
      throw new RuntimeException("Unknown block cache version [" + blockCacheVersion + "] can be [v1,v2]");
    }
    LOG.info("Shard Server using index [{0}] bind address [{1}]", serverIndex, bindAddress + ":" + instanceBindPort);

    String nodeNameHostName = getNodeName(configuration, BLUR_SHARD_HOSTNAME);
    String nodeName = nodeNameHostName + ":" + instanceBindPort;
    configuration.set(BLUR_NODENAME, nodeName);

    BlurQueryChecker queryChecker = new BlurQueryChecker(configuration);

    String cluster = configuration.get(BLUR_CLUSTER_NAME, BLUR_CLUSTER);

    final ZooKeeper zooKeeper = setupZookeeper(configuration, cluster);
    final ZookeeperClusterStatus clusterStatus = new ZookeeperClusterStatus(zooKeeper, configuration, config);

    BlurFilterCache filterCache = getFilterCache(configuration);

    DistributedLayoutFactory distributedLayoutFactory = DistributedLayoutFactoryImpl.getDistributedLayoutFactory(
        configuration, cluster, zooKeeper);

    long requestCacheSize = configuration.getLong(BLUR_SHARD_REQUEST_CACHE_SIZE, 10000000);
    final ThriftCache thriftCache = new ThriftCache(requestCacheSize);

    long safeModeDelay = configuration.getLong(BLUR_SHARD_SAFEMODEDELAY, 60000);
    int shardOpenerThreadCount = configuration.getInt(BLUR_SHARD_OPENER_THREAD_COUNT, 16);
    int maxMergeThreads = configuration.getInt(BLUR_SHARD_MERGE_THREAD_COUNT, 3);
    int minimumNumberOfNodesBeforeExitingSafeMode = configuration.getInt(
        BLUR_SHARD_SERVER_MINIMUM_BEFORE_SAFEMODE_EXIT, 0);
    int internalSearchThreads = configuration.getInt(BLUR_SHARD_INTERNAL_SEARCH_THREAD_COUNT, 16);
    final Timer hdfsKeyValueTimer = new Timer("HDFS KV Store", true);
    final Timer indexImporterTimer = new Timer("IndexImporter", true);
    final Timer indexBulkTimer = new Timer("BulkIndex", true);
    final Timer indexIdleWriterTimer = new Timer("BlurIdleWriter", true);
    long smallMergeThreshold = configuration.getLong(BLUR_SHARD_SMALL_MERGE_THRESHOLD, 128 * 1000 * 1000);
    long maxWriterIdle = configuration.getLong(BLUR_SHARD_INDEX_MAX_IDLE_TIME, 30000);
    SequentialReadControl sequentialReadControl = new SequentialReadControl(configuration);
    final DistributedIndexServer indexServer = new DistributedIndexServer(config, zooKeeper, clusterStatus,
        filterCache, blockCacheDirectoryFactory, distributedLayoutFactory, cluster, nodeName, safeModeDelay,
        shardOpenerThreadCount, maxMergeThreads, internalSearchThreads, minimumNumberOfNodesBeforeExitingSafeMode,
        hdfsKeyValueTimer, indexImporterTimer, smallMergeThreshold, indexBulkTimer, thriftCache, sequentialReadControl,
        indexIdleWriterTimer, maxWriterIdle);

    BooleanQuery.setMaxClauseCount(configuration.getInt(BLUR_MAX_CLAUSE_COUNT, 1024));

    int maxHeapPerRowFetch = configuration.getInt(BLUR_MAX_HEAP_PER_ROW_FETCH, 10000000);
    int remoteFetchCount = configuration.getInt(BLUR_CONTROLLER_REMOTE_FETCH_COUNT, 100);
    int fetchCount = configuration.getInt(BLUR_SHARD_FETCHCOUNT, 110);
    if (fetchCount + 1 <= remoteFetchCount) {
      LOG.warn("[" + BLUR_SHARD_FETCHCOUNT + "] [" + fetchCount + "] is equal to or less than ["
          + BLUR_CONTROLLER_REMOTE_FETCH_COUNT + "] [" + remoteFetchCount + "], should be at least 1 greater.");
    }
    int indexManagerThreadCount = configuration.getInt(BLUR_INDEXMANAGER_SEARCH_THREAD_COUNT, 32);
    int mutateThreadCount = configuration.getInt(BLUR_INDEXMANAGER_MUTATE_THREAD_COUNT, 32);
    int facetThreadCount = configuration.getInt(BLUR_INDEXMANAGER_FACET_THREAD_COUNT, 16);
    long statusCleanupTimerDelay = TimeUnit.SECONDS.toMillis(10);
    int cacheSize = configuration.getInt(BLUR_SHARD_DEEP_PAGING_CACHE_SIZE, 1000);
    DeepPagingCache deepPagingCache = new DeepPagingCache(cacheSize);

    MemoryAllocationWatcher memoryAllocationWatcher = new MemoryAllocationWatcher() {
      @Override
      public <T, E extends Exception> T run(Watcher<T, E> w) throws E {
        return w.run();
      }
    };

    QueryStatusManager statusManager = new QueryStatusManager(statusCleanupTimerDelay);

    final IndexManager indexManager = new IndexManager(indexServer, clusterStatus, filterCache, maxHeapPerRowFetch,
        fetchCount, indexManagerThreadCount, mutateThreadCount, facetThreadCount, deepPagingCache,
        memoryAllocationWatcher, statusManager);

    File tmpPath = getTmpPath(configuration);
    int numberOfShardWorkerCommandThreads = configuration.getInt(BLUR_SHARD_COMMAND_WORKER_THREADS, 16);
    int numberOfShardDriverCommandThreads = configuration.getInt(BLUR_SHARD_COMMAND_DRIVER_THREADS, 16);
    String commandPath = configuration.get(BLUR_COMMAND_LIB_PATH, getCommandLibPath());
    if (commandPath != null) {
      LOG.info("Command Path was set to [{0}].", commandPath);
    } else {
      LOG.info("Command Path was not set.");
    }
    final ShardCommandManager commandManager = new ShardCommandManager(indexServer, tmpPath, commandPath,
        numberOfShardWorkerCommandThreads, numberOfShardDriverCommandThreads, Connection.DEFAULT_TIMEOUT, config,
        nodeName);

    clusterStatus.registerActionOnTableStateChange(new Action() {
      @Override
      public void action() {
        thriftCache.clear();
      }
    });

    final BlurShardServer shardServer = new BlurShardServer();
    shardServer.setNodeName(nodeName);
    shardServer.setCommandManager(commandManager);
    shardServer.setIndexServer(indexServer);
    shardServer.setIndexManager(indexManager);
    shardServer.setZookeeper(zooKeeper);
    shardServer.setClusterStatus(clusterStatus);
    shardServer.setQueryChecker(queryChecker);
    shardServer.setMaxRecordsPerRowFetchRequest(configuration.getInt(BLUR_MAX_RECORDS_PER_ROW_FETCH_REQUEST, 1000));
    shardServer.setConfiguration(configuration);
    shardServer.init();

    final TraceStorage traceStorage = setupTraceStorage(configuration, config);
    Trace.setStorage(traceStorage);
    Trace.setNodeName(nodeName);

    List<ServerSecurityFilter> serverSecurity = getServerSecurityList(configuration,
        ServerSecurityFilterFactory.ServerType.SHARD);

    Iface iface = new ThriftCacheServer(configuration, shardServer, indexServer, thriftCache);
    iface = BlurUtil.wrapFilteredBlurServer(configuration, iface, true);
    iface = ServerSecurityUtil.applySecurity(iface, serverSecurity, true);
    iface = BlurUtil.recordMethodCallsAndAverageTimes(iface, Iface.class, false);
    iface = BlurUtil.runWithUser(iface, false);
    iface = BlurUtil.runTrace(iface, false);
    iface = BlurUtil.lastChanceErrorHandling(iface, Iface.class);

    int configGuiPort = Integer.parseInt(configuration.get(BLUR_GUI_SHARD_PORT));
    int instanceGuiPort = configGuiPort + serverIndex;

    if (configGuiPort == 0) {
      instanceGuiPort = 0;
    }

    StreamServer streamServer;
    int streamThreadCount = configuration.getInt(BLUR_STREAM_SERVER_THREADS, 100);
    if (streamThreadCount > 0) {
      StreamProcessor streamProcessor = new StreamProcessor(indexServer, tmpPath);
      streamServer = new StreamServer(0, streamThreadCount, streamProcessor);
      streamServer.start();
      configuration.setInt(BLUR_STREAM_SERVER_RUNNING_PORT, streamServer.getPort());
      LOG.info("Stream server started on port [{0}]", streamServer.getPort());
    } else {
      streamServer = null;
    }

    final HttpJettyServer httpServer;
    if (configGuiPort >= 0) {
      httpServer = new HttpJettyServer(HttpJettyServer.class, instanceGuiPort);
      int port = httpServer.getLocalPort();
      configuration.setInt(BLUR_HTTP_STATUS_RUNNING_PORT, port);
    } else {
      httpServer = null;
    }
    if (httpServer != null) {
      WebAppContext context = httpServer.getContext();
      context.addServlet(new ServletHolder(new TServlet(new Blur.Processor<Blur.Iface>(iface),
          new TJSONProtocol.Factory())), "/blur");
      if (enableJsonReporter) {
        JSONReporter.enable("json-reporter", 1, TimeUnit.SECONDS, 60);
      }
    }

    int threadCount = configuration.getInt(BLUR_SHARD_SERVER_THRIFT_THREAD_COUNT, 32);

    ShardServerEventHandler eventHandler = new ShardServerEventHandler();

    final ThriftBlurShardServer server = new ThriftBlurShardServer();
    server.setNodeName(nodeName);
    server.setServerTransport(serverTransport);
    server.setThreadCount(threadCount);
    server.setIface(iface);
    server.setEventHandler(eventHandler);
    server.setAcceptQueueSizePerThread(configuration.getInt(BLUR_SHARD_THRIFT_ACCEPT_QUEUE_SIZE_PER_THREAD, 4));
    server.setMaxReadBufferBytes(configuration.getLong(BLUR_SHARD_THRIFT_MAX_READ_BUFFER_BYTES, Long.MAX_VALUE));
    server.setSelectorThreads(configuration.getInt(BLUR_SHARD_THRIFT_SELECTOR_THREADS, 2));
    server.setMaxFrameSize(configuration.getInt(BLUR_THRIFT_MAX_FRAME_SIZE, BLUR_THRIFT_DEFAULT_MAX_FRAME_SIZE));
    server.setConfiguration(configuration);

    // This will shutdown the server when the correct path is set in zk
    BlurShutdown shutdown = new BlurShutdown() {
      @Override
      public void shutdown() {
        ThreadWatcher threadWatcher = ThreadWatcher.instance();
        quietClose(streamServer, makeCloseable(hdfsKeyValueTimer), makeCloseable(indexImporterTimer),
            makeCloseable(indexBulkTimer), blockCacheDirectoryFactory, commandManager, traceStorage, server,
            shardServer, indexManager, statusManager, indexServer, threadWatcher, clusterStatus, zooKeeper, httpServer);
      }
    };
    server.setShutdown(shutdown);
    new BlurServerShutDown().register(shutdown, zooKeeper);
    return server;
  }

  protected static Closeable makeCloseable(final Timer timer) {
    return new CloseableTimer(timer);
  }

  @SuppressWarnings("unchecked")
  private static BlurFilterCache getFilterCache(BlurConfiguration configuration) {
    String blurFilterCacheClass = configuration.get(BLUR_SHARD_FILTER_CACHE_CLASS);
    if (blurFilterCacheClass != null) {
      try {
        Class<? extends BlurFilterCache> clazz = (Class<? extends BlurFilterCache>) Class.forName(blurFilterCacheClass);
        Class<?>[] types = new Class<?>[] { BlurConfiguration.class };
        Constructor<? extends BlurFilterCache> constructor = clazz.getConstructor(types);
        return constructor.newInstance(new Object[] { configuration });
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    }
    return new DefaultBlurFilterCache(configuration);
  }

}