<commit_before>package mesosphere.marathon.health

import mesosphere.marathon.state.Timestamp
import mesosphere.marathon.Protos.MarathonTask
import mesosphere.marathon.Protos.HealthCheckDefinition.Protocol.{HTTP, TCP}

import akka.actor.{Actor, ActorLogging}
import akka.util.Timeout

import spray.http._
import spray.client.pipelining._

import scala.collection.JavaConverters._
import scala.concurrent.Future
import scala.util.{Try, Success, Failure}

import java.net.{Socket, InetSocketAddress}


class HealthCheckWorkerActor extends Actor with ActorLogging {

  import HealthCheckWorker._

  implicit val system = context.system
  import context.dispatcher // execution context for futures

  def receive = {
    case HealthCheckJob(task, check) => {
      val replyTo = sender // avoids closing over the volatile sender ref

      val replyWithHealth = doCheck(task, check)

      replyWithHealth.onComplete {
        case Success(result) => replyTo ! result
        case Failure(t) => replyTo ! Unhealthy(
          task.getId,
          Timestamp.now,
          s"${t.getClass.getSimpleName}: ${t.getMessage}"
        )
      }

      replyWithHealth andThen { case _ => context stop self }
    }
  }

  def doCheck(task: MarathonTask, check: HealthCheck): Future[HealthResult] =
    Try(task.getPortsList.asScala(check.portIndex)).toOption match {
      case None =>
        Future { Unhealthy(task.getId, Timestamp.now, "Invalid port index") }

      case Some(port) => check.protocol match {
        case HTTP => http(task, check, port)
        case TCP  => tcp(task, check, port)
      }
    }

  def http(task: MarathonTask, check: HealthCheck, port: Int): Future[HealthResult] = {
    val host = task.getHost
    val rawPath = check.path.getOrElse("")
    val absolutePath = if (rawPath.startsWith("/")) rawPath else s"/$rawPath"
    val url = s"http://${host}:${port}${absolutePath}"
    log.debug("Checking the health of [{}] via HTTP", url)

    def get(url: String): Future[HttpResponse] = {
      implicit val requestTimeout = Timeout(check.timeout)
      val pipeline: HttpRequest => Future[HttpResponse] = sendReceive
      pipeline(Get(url))
    }

    get(url).map { response =>
      if (acceptableResponses contains response.status.intValue)
        Healthy(task.getId, Timestamp.now)
      else
        Unhealthy(task.getId, Timestamp.now, response.status.toString)
    }
  }

  def tcp(task: MarathonTask, check: HealthCheck, port: Int): Future[HealthResult] = {
    val host = task.getHost
    val address = s"${host}:${port}"
    val timeoutMillis = check.timeout.toMillis.toInt
    log.debug("Checking the health of [{}] via TCP", address)

    Future {
      val address = InetSocketAddress.createUnresolved(host, port)
      val socket = new Socket
      socket.connect(address, timeoutMillis)
      socket.close
      Healthy(task.getId, Timestamp.now)
    }
  }

}

object HealthCheckWorker {

  // Similar to AWS R53, we accept all responses in [200, 399]
  protected[health] val acceptableResponses = (200 until 399).toSet

  case class HealthCheckJob(task: MarathonTask, check: HealthCheck)

  sealed trait HealthResult {
    val taskId: String
    val time: Timestamp
  }

  case class Healthy(
    taskId: String,
    time: Timestamp = Timestamp.now
  ) extends HealthResult

  case class Unhealthy(
    taskId: String,
    time: Timestamp = Timestamp.now,
    cause: String
  ) extends HealthResult

}<commit_msg>Use correct range<commit_after>package mesosphere.marathon.health

import mesosphere.marathon.state.Timestamp
import mesosphere.marathon.Protos.MarathonTask
import mesosphere.marathon.Protos.HealthCheckDefinition.Protocol.{HTTP, TCP}

import akka.actor.{Actor, ActorLogging}
import akka.util.Timeout

import spray.http._
import spray.client.pipelining._

import scala.collection.JavaConverters._
import scala.concurrent.Future
import scala.util.{Try, Success, Failure}

import java.net.{Socket, InetSocketAddress}


class HealthCheckWorkerActor extends Actor with ActorLogging {

  import HealthCheckWorker._

  implicit val system = context.system
  import context.dispatcher // execution context for futures

  def receive = {
    case HealthCheckJob(task, check) => {
      val replyTo = sender // avoids closing over the volatile sender ref

      val replyWithHealth = doCheck(task, check)

      replyWithHealth.onComplete {
        case Success(result) => replyTo ! result
        case Failure(t) => replyTo ! Unhealthy(
          task.getId,
          Timestamp.now,
          s"${t.getClass.getSimpleName}: ${t.getMessage}"
        )
      }

      replyWithHealth andThen { case _ => context stop self }
    }
  }

  def doCheck(task: MarathonTask, check: HealthCheck): Future[HealthResult] =
    Try(task.getPortsList.asScala(check.portIndex)).toOption match {
      case None =>
        Future { Unhealthy(task.getId, Timestamp.now, "Invalid port index") }

      case Some(port) => check.protocol match {
        case HTTP => http(task, check, port)
        case TCP  => tcp(task, check, port)
      }
    }

  def http(task: MarathonTask, check: HealthCheck, port: Int): Future[HealthResult] = {
    val host = task.getHost
    val rawPath = check.path.getOrElse("")
    val absolutePath = if (rawPath.startsWith("/")) rawPath else s"/$rawPath"
    val url = s"http://${host}:${port}${absolutePath}"
    log.debug("Checking the health of [{}] via HTTP", url)

    def get(url: String): Future[HttpResponse] = {
      implicit val requestTimeout = Timeout(check.timeout)
      val pipeline: HttpRequest => Future[HttpResponse] = sendReceive
      pipeline(Get(url))
    }

    get(url).map { response =>
      if (acceptableResponses contains response.status.intValue)
        Healthy(task.getId, Timestamp.now)
      else
        Unhealthy(task.getId, Timestamp.now, response.status.toString)
    }
  }

  def tcp(task: MarathonTask, check: HealthCheck, port: Int): Future[HealthResult] = {
    val host = task.getHost
    val address = s"${host}:${port}"
    val timeoutMillis = check.timeout.toMillis.toInt
    log.debug("Checking the health of [{}] via TCP", address)

    Future {
      val address = InetSocketAddress.createUnresolved(host, port)
      val socket = new Socket
      socket.connect(address, timeoutMillis)
      socket.close
      Healthy(task.getId, Timestamp.now)
    }
  }

}

object HealthCheckWorker {

  // Similar to AWS R53, we accept all responses in [200, 399]
  protected[health] val acceptableResponses = Range(200, 400)

  case class HealthCheckJob(task: MarathonTask, check: HealthCheck)

  sealed trait HealthResult {
    val taskId: String
    val time: Timestamp
  }

  case class Healthy(
    taskId: String,
    time: Timestamp = Timestamp.now
  ) extends HealthResult

  case class Unhealthy(
    taskId: String,
    time: Timestamp = Timestamp.now,
    cause: String
  ) extends HealthResult

}
{'fraction_non_alphanumeric': 0.1046073601854535, 'fraction_numerical': 0.004056795131845842, 'mean_word_length': 3.3333333333333335, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '6080044', 'n_tokens_mistral': 2114, 'n_tokens_neox': 2028, 'n_words': 533}