<commit_before>/*
 *  ____    ____    _____    ____    ___     ____ 
 * |  _ \  |  _ \  | ____|  / ___|  / _/    / ___|        Precog (R)
 * | |_) | | |_) | |  _|   | |     | |  /| | |  _         Advanced Analytics Engine for NoSQL Data
 * |  __/  |  _ <  | |___  | |___  |/ _| | | |_| |        Copyright (C) 2010 - 2013 SlamData, Inc.
 * |_|     |_| \_\ |_____|  \____|   /__/   \____|        All Rights Reserved.
 *
 * This program is free software: you can redistribute it and/or modify it under the terms of the 
 * GNU Affero General Public License as published by the Free Software Foundation, either version 
 * 3 of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; 
 * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See 
 * the GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License along with this 
 * program. If not, see <http://www.gnu.org/licenses/>.
 *
 */
package com.precog.yggdrasil
package actor

import metadata.ColumnMetadata
import com.precog.accounts.BasicAccountManager
import com.precog.util._
import com.precog.common._
import com.precog.common.kafka._

import akka.actor._
import akka.dispatch._
import akka.util._
import akka.util.duration._
import akka.pattern.ask
import akka.pattern.gracefulStop

import blueeyes.json._

import com.weiglewilczek.slf4s.Logging
import org.streum.configrity.converter.Extra._
import _root_.kafka.consumer._

import java.io.File
import java.net.InetAddress

import scalaz._
import scalaz.syntax.id._


trait KafkaIngestActorProjectionSystemConfig extends ShardConfig {
  case class IngestConfig(
    bufferSize: Int,
    maxParallel: Int, 
    batchTimeout: Timeout, 
    failureLogRoot: File,
    maxConsecutiveFailures: Int)

  def kafkaHost: String = config[String]("kafka.batch.host")
  def kafkaPort: Int = config[Int]("kafka.batch.port")
  def kafkaTopic: String = config[String]("kafka.batch.topic") 
  def kafkaSocketTimeout: Duration = config[Long]("kafka.socket_timeout", 5000) millis
  def kafkaBufferSize: Int = config[Int]("kafka.buffer_size", 64 * 1024)

  def ingestConfig = config.detach("ingest") |> { config =>
    for {
      failureLogRoot <- config.get[File]("failure_log_root") if config[Boolean]("enabled", false)
    } yield {
      IngestConfig(
        bufferSize = config[Int]("buffer_size", 1024 * 1024),
        maxParallel = config[Int]("max_parallel", 5),
        batchTimeout = config[Int]("timeout", 120) seconds,
        maxConsecutiveFailures = config[Int]("ingest.max_consecutive_failures", 3),
        failureLogRoot = failureLogRoot)
    }
  }

  def zookeeperHosts: String = config[String]("zookeeper.hosts")
  def zookeeperBase: List[String] = config[List[String]]("zookeeper.basepath")
  def zookeeperPrefix: String = config[String]("zookeeper.prefix")   

  def serviceUID: ServiceUID = ZookeeperSystemCoordination.extractServiceUID(config)

  lazy val shardId = {
    val suid = serviceUID
    serviceUID.hostId + serviceUID.serviceId
  }

  val logPrefix = "[Production Yggdrasil Shard]"
}

trait KafkaIngestActorProjectionSystem extends ShardSystemActorModule {
  type YggConfig <: KafkaIngestActorProjectionSystemConfig

  def ingestFailureLog(checkpoint: YggCheckpoint): IngestFailureLog

  override def initIngestActor(actorSystem: ActorSystem, checkpoint: YggCheckpoint, metadataActor: ActorRef, accountManager: BasicAccountManager[Future]) = {
    val consumer = new SimpleConsumer(yggConfig.kafkaHost, yggConfig.kafkaPort, yggConfig.kafkaSocketTimeout.toMillis.toInt, yggConfig.kafkaBufferSize)
    yggConfig.ingestConfig map { conf => actorSystem.actorOf(Props(
      new KafkaShardIngestActor(shardId = yggConfig.shardId, 
                                         initialCheckpoint = checkpoint, 
                                         consumer = consumer, 
                                         topic = yggConfig.kafkaTopic, 
                                         accountManager = accountManager,
                                         ingestFailureLog = ingestFailureLog(checkpoint),
                                         fetchBufferSize = conf.bufferSize,
                                         ingestTimeout = conf.batchTimeout,
                                         maxCacheSize = conf.maxParallel,
                                         maxConsecutiveFailures = conf.maxConsecutiveFailures) {
        def handleBatchComplete(pendingCheckpoint: YggCheckpoint, updates: Seq[(ProjectionDescriptor, Option[ColumnMetadata])]) {
          logger.debug(pendingCheckpoint + " to be updated")
          metadataActor ! IngestBatchMetadata(updates, pendingCheckpoint.messageClock, Some(pendingCheckpoint.offset))
        }
      }), "ingest")
    }
  }

  override def checkpointCoordination = ZookeeperSystemCoordination(yggConfig.zookeeperHosts, yggConfig.serviceUID, yggConfig.ingestConfig.isDefined) 
}

trait StandaloneShardSystemConfig extends ShardConfig {
  def shardId = "standalone"
  def logPrefix = "[Standalone Yggdrasil Shard]"
}

trait StandaloneActorProjectionSystem extends ShardSystemActorModule {
  type YggConfig <: StandaloneShardSystemConfig
  override def initIngestActor(actorSystem: ActorSystem, checkpoint: YggCheckpoint, metadataActor: ActorRef, accountManager: BasicAccountManager[Future]) = None
  override def checkpointCoordination = CheckpointCoordination.Noop
}

// vim: set ts=4 sw=4 et:
/* tmux
type ShardSystemActorConfigs */

<commit_msg>Only create the zookeeper consumer if we're going to have ingest online.<commit_after>/*
 *  ____    ____    _____    ____    ___     ____ 
 * |  _ \  |  _ \  | ____|  / ___|  / _/    / ___|        Precog (R)
 * | |_) | | |_) | |  _|   | |     | |  /| | |  _         Advanced Analytics Engine for NoSQL Data
 * |  __/  |  _ <  | |___  | |___  |/ _| | | |_| |        Copyright (C) 2010 - 2013 SlamData, Inc.
 * |_|     |_| \_\ |_____|  \____|   /__/   \____|        All Rights Reserved.
 *
 * This program is free software: you can redistribute it and/or modify it under the terms of the 
 * GNU Affero General Public License as published by the Free Software Foundation, either version 
 * 3 of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; 
 * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See 
 * the GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License along with this 
 * program. If not, see <http://www.gnu.org/licenses/>.
 *
 */
package com.precog.yggdrasil
package actor

import metadata.ColumnMetadata
import com.precog.accounts.BasicAccountManager
import com.precog.util._
import com.precog.common._
import com.precog.common.kafka._

import akka.actor._
import akka.dispatch._
import akka.util._
import akka.util.duration._
import akka.pattern.ask
import akka.pattern.gracefulStop

import blueeyes.json._

import com.weiglewilczek.slf4s.Logging
import org.streum.configrity.converter.Extra._
import _root_.kafka.consumer._

import java.io.File
import java.net.InetAddress

import scalaz._
import scalaz.syntax.id._


trait KafkaIngestActorProjectionSystemConfig extends ShardConfig {
  case class IngestConfig(
    bufferSize: Int,
    maxParallel: Int, 
    batchTimeout: Timeout, 
    failureLogRoot: File,
    maxConsecutiveFailures: Int)

  def kafkaHost: String = config[String]("kafka.batch.host")
  def kafkaPort: Int = config[Int]("kafka.batch.port")
  def kafkaTopic: String = config[String]("kafka.batch.topic") 
  def kafkaSocketTimeout: Duration = config[Long]("kafka.socket_timeout", 5000) millis
  def kafkaBufferSize: Int = config[Int]("kafka.buffer_size", 64 * 1024)

  def ingestConfig = config.detach("ingest") |> { config =>
    for {
      failureLogRoot <- config.get[File]("failure_log_root") if config[Boolean]("enabled", false)
    } yield {
      IngestConfig(
        bufferSize = config[Int]("buffer_size", 1024 * 1024),
        maxParallel = config[Int]("max_parallel", 5),
        batchTimeout = config[Int]("timeout", 120) seconds,
        maxConsecutiveFailures = config[Int]("ingest.max_consecutive_failures", 3),
        failureLogRoot = failureLogRoot)
    }
  }

  def zookeeperHosts: String = config[String]("zookeeper.hosts")
  //def zookeeperBase: List[String] = config[List[String]]("zookeeper.basepath")
  //def zookeeperPrefix: String = config[String]("zookeeper.prefix")   

  def serviceUID: ServiceUID = ZookeeperSystemCoordination.extractServiceUID(config)

  lazy val shardId = {
    val suid = serviceUID
    serviceUID.hostId + serviceUID.serviceId
  }

  val logPrefix = "[Production Yggdrasil Shard]"
}

trait KafkaIngestActorProjectionSystem extends ShardSystemActorModule {
  type YggConfig <: KafkaIngestActorProjectionSystemConfig

  def ingestFailureLog(checkpoint: YggCheckpoint): IngestFailureLog

  override def initIngestActor(actorSystem: ActorSystem, checkpoint: YggCheckpoint, metadataActor: ActorRef, accountManager: BasicAccountManager[Future]) = {
    yggConfig.ingestConfig map { conf => 
      val consumer = new SimpleConsumer(yggConfig.kafkaHost, 
                                        yggConfig.kafkaPort, 
                                        yggConfig.kafkaSocketTimeout.toMillis.toInt, 
                                        yggConfig.kafkaBufferSize)

      actorSystem.actorOf(Props(
        new KafkaShardIngestActor( shardId = yggConfig.shardId, 
                                   initialCheckpoint = checkpoint, 
                                   consumer = consumer, 
                                   topic = yggConfig.kafkaTopic, 
                                   accountManager = accountManager,
                                   ingestFailureLog = ingestFailureLog(checkpoint),
                                   fetchBufferSize = conf.bufferSize,
                                   ingestTimeout = conf.batchTimeout,
                                   maxCacheSize = conf.maxParallel,
                                   maxConsecutiveFailures = conf.maxConsecutiveFailures) {

        def handleBatchComplete(ck: YggCheckpoint, updates: Seq[(ProjectionDescriptor, Option[ColumnMetadata])]) {
          logger.debug(ck + " to be updated")
          metadataActor ! IngestBatchMetadata(updates, ck.messageClock, Some(ck.offset))
        }
      }), "ingest")
    }
  }

  override def checkpointCoordination = ZookeeperSystemCoordination(yggConfig.zookeeperHosts, yggConfig.serviceUID, yggConfig.ingestConfig.isDefined) 
}

trait StandaloneShardSystemConfig extends ShardConfig {
  def shardId = "standalone"
  def logPrefix = "[Standalone Yggdrasil Shard]"
}

trait StandaloneActorProjectionSystem extends ShardSystemActorModule {
  type YggConfig <: StandaloneShardSystemConfig
  override def initIngestActor(actorSystem: ActorSystem, checkpoint: YggCheckpoint, metadataActor: ActorRef, accountManager: BasicAccountManager[Future]) = None
  override def checkpointCoordination = CheckpointCoordination.Noop
}

// vim: set ts=4 sw=4 et:
/* tmux
type ShardSystemActorConfigs */


{'fraction_non_alphanumeric': 0.08949691085613416, 'fraction_numerical': 0.00617828773168579, 'mean_word_length': 3.0180851063829786, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 32, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1922347', 'n_tokens_mistral': 3382, 'n_tokens_neox': 3222, 'n_words': 764}