<commit_before>// Copyright (c) 2016-2017 Chef Software Inc. and/or applicable contributors
//
// 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.

use std::fmt;
use std::fs::File;
use std::io::{self, Read};
use std::net::{IpAddr, Ipv4Addr, SocketAddr, SocketAddrV4, ToSocketAddrs};
use std::ops::{Deref, DerefMut};
use std::option;
use std::result;
use std::str::FromStr;
use std::sync::{Arc, Condvar, Mutex, RwLock};
use std::thread;

use actix;
use actix_web::{
    http::StatusCode, pred::Predicate, server, App, FromRequest, HttpRequest, HttpResponse, Path,
    Request,
};
use hcore::service::ServiceGroup;
use protocol::socket_addr_env_or_default;
use serde_json::{self, Value as Json};

use error::{Error, Result, SupError};
use manager;
use manager::service::hooks::{self, HealthCheckHook};
use manager::service::HealthCheck;

use feat;

static LOGKEY: &'static str = "HG";
const APIDOCS: &'static str = include_str!(concat!(env!("OUT_DIR"), "/api.html"));

/// Default listening port for the HTTPGateway listener.
pub const DEFAULT_PORT: u16 = 9631;

/// Default environment variable override for HTTPGateway listener address.
pub const DEFAULT_ADDRESS_ENVVAR: &'static str = "HAB_LISTEN_HTTP";

#[derive(PartialEq, Eq, Debug, Clone)]
pub struct ListenAddr(SocketAddr);

impl ListenAddr {
    pub fn new(ip: IpAddr, port: u16) -> ListenAddr {
        ListenAddr(SocketAddr::new(ip, port))
    }
}

impl Default for ListenAddr {
    fn default() -> ListenAddr {
        ListenAddr(socket_addr_env_or_default(
            DEFAULT_ADDRESS_ENVVAR,
            SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(0, 0, 0, 0), DEFAULT_PORT)),
        ))
    }
}

impl Deref for ListenAddr {
    type Target = SocketAddr;

    fn deref(&self) -> &SocketAddr {
        &self.0
    }
}

impl DerefMut for ListenAddr {
    fn deref_mut(&mut self) -> &mut SocketAddr {
        &mut self.0
    }
}

impl FromStr for ListenAddr {
    type Err = SupError;

    fn from_str(val: &str) -> Result<Self> {
        match SocketAddr::from_str(val) {
            Ok(addr) => Ok(ListenAddr(addr)),
            Err(_) => match IpAddr::from_str(val) {
                Ok(ip) => {
                    let mut addr = ListenAddr::default();
                    addr.set_ip(ip);
                    Ok(addr)
                }
                Err(_) => Err(sup_error!(Error::IPFailed)),
            },
        }
    }
}

impl ToSocketAddrs for ListenAddr {
    type Iter = option::IntoIter<SocketAddr>;

    fn to_socket_addrs(&self) -> io::Result<Self::Iter> {
        self.0.to_socket_addrs()
    }
}

impl fmt::Display for ListenAddr {
    fn fmt(&self, f: &mut fmt::Formatter) -> result::Result<(), fmt::Error> {
        write!(f, "{}", self.0)
    }
}

#[derive(Default, Serialize)]
struct HealthCheckBody {
    status: String,
    stdout: String,
    stderr: String,
}

impl Into<StatusCode> for HealthCheck {
    fn into(self) -> StatusCode {
        match self {
            HealthCheck::Ok | HealthCheck::Warning => StatusCode::OK,
            HealthCheck::Critical => StatusCode::SERVICE_UNAVAILABLE,
            HealthCheck::Unknown => StatusCode::INTERNAL_SERVER_ERROR,
        }
    }
}

struct AppState {
    gateway_state: Arc<RwLock<manager::GatewayState>>,
}

impl AppState {
    fn new(gs: Arc<RwLock<manager::GatewayState>>) -> Self {
        AppState { gateway_state: gs }
    }
}

#[derive(Debug, PartialEq, Eq)]
pub enum ServerStartup {
    NotStarted,
    Started,
    BindFailed,
}

pub struct Server;

impl Server {
    pub fn run(
        listen_addr: ListenAddr,
        gateway_state: Arc<RwLock<manager::GatewayState>>,
        control: Arc<(Mutex<ServerStartup>, Condvar)>,
    ) {
        thread::spawn(move || {
            let &(ref lock, ref cvar) = &*control;
            let sys = actix::System::new("sup-http-gateway");

            let bind = server::new(move || {
                let app_state = AppState::new(gateway_state.clone());
                App::with_state(app_state).configure(routes)
            }).workers(2)
            .bind(listen_addr.to_string());

            // We need to create this scope on purpose here because if we don't, the lock never
            // releases, and the supervisor will wait forever on cvar. Creating this artifical
            // scope forces the lock to release, and things work as expected.
            {
                let mut started = lock.lock().expect("Control mutex is poisoned");

                *started = match bind {
                    Ok(b) => {
                        b.start();
                        ServerStartup::Started
                    }
                    Err(e) => {
                        error!("HTTP gateway failed to bind: {:?}", e);
                        ServerStartup::BindFailed
                    }
                };
            }

            cvar.notify_one();
            sys.run();
        });
    }
}

struct RedactHTTP;<commit_msg>Tune the HTTP threads via HAB_SUP_HTTP_THREADS<commit_after>// Copyright (c) 2016-2017 Chef Software Inc. and/or applicable contributors
//
// 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.

use std::{
    fmt,
    fs::File,
    io::{self, Read},
    net::{IpAddr, Ipv4Addr, SocketAddr, SocketAddrV4, ToSocketAddrs},
    ops::{Deref, DerefMut},
    option, result,
    str::FromStr,
    sync::{Arc, Condvar, Mutex, RwLock},
    thread,
};

use actix;
use actix_web::{
    http::StatusCode, pred::Predicate, server, App, FromRequest, HttpRequest, HttpResponse, Path,
    Request,
};
use hcore::{env as henv, service::ServiceGroup};
use protocol::socket_addr_env_or_default;
use serde_json::{self, Value as Json};

use error::{Error, Result, SupError};
use manager;
use manager::service::hooks::{self, HealthCheckHook};
use manager::service::HealthCheck;

use feat;

static LOGKEY: &'static str = "HG";
const APIDOCS: &'static str = include_str!(concat!(env!("OUT_DIR"), "/api.html"));
pub const HTTP_THREADS_ENVVAR: &'static str = "HAB_SUP_HTTP_THREADS";
pub const HTTP_THREAD_COUNT: usize = 2;

/// Default listening port for the HTTPGateway listener.
pub const DEFAULT_PORT: u16 = 9631;

/// Default environment variable override for HTTPGateway listener address.
pub const DEFAULT_ADDRESS_ENVVAR: &'static str = "HAB_LISTEN_HTTP";

#[derive(PartialEq, Eq, Debug, Clone)]
pub struct ListenAddr(SocketAddr);

impl ListenAddr {
    pub fn new(ip: IpAddr, port: u16) -> ListenAddr {
        ListenAddr(SocketAddr::new(ip, port))
    }
}

impl Default for ListenAddr {
    fn default() -> ListenAddr {
        ListenAddr(socket_addr_env_or_default(
            DEFAULT_ADDRESS_ENVVAR,
            SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(0, 0, 0, 0), DEFAULT_PORT)),
        ))
    }
}

impl Deref for ListenAddr {
    type Target = SocketAddr;

    fn deref(&self) -> &SocketAddr {
        &self.0
    }
}

impl DerefMut for ListenAddr {
    fn deref_mut(&mut self) -> &mut SocketAddr {
        &mut self.0
    }
}

impl FromStr for ListenAddr {
    type Err = SupError;

    fn from_str(val: &str) -> Result<Self> {
        match SocketAddr::from_str(val) {
            Ok(addr) => Ok(ListenAddr(addr)),
            Err(_) => match IpAddr::from_str(val) {
                Ok(ip) => {
                    let mut addr = ListenAddr::default();
                    addr.set_ip(ip);
                    Ok(addr)
                }
                Err(_) => Err(sup_error!(Error::IPFailed)),
            },
        }
    }
}

impl ToSocketAddrs for ListenAddr {
    type Iter = option::IntoIter<SocketAddr>;

    fn to_socket_addrs(&self) -> io::Result<Self::Iter> {
        self.0.to_socket_addrs()
    }
}

impl fmt::Display for ListenAddr {
    fn fmt(&self, f: &mut fmt::Formatter) -> result::Result<(), fmt::Error> {
        write!(f, "{}", self.0)
    }
}

#[derive(Default, Serialize)]
struct HealthCheckBody {
    status: String,
    stdout: String,
    stderr: String,
}

impl Into<StatusCode> for HealthCheck {
    fn into(self) -> StatusCode {
        match self {
            HealthCheck::Ok | HealthCheck::Warning => StatusCode::OK,
            HealthCheck::Critical => StatusCode::SERVICE_UNAVAILABLE,
            HealthCheck::Unknown => StatusCode::INTERNAL_SERVER_ERROR,
        }
    }
}

struct AppState {
    gateway_state: Arc<RwLock<manager::GatewayState>>,
}

impl AppState {
    fn new(gs: Arc<RwLock<manager::GatewayState>>) -> Self {
        AppState { gateway_state: gs }
    }
}

#[derive(Debug, PartialEq, Eq)]
pub enum ServerStartup {
    NotStarted,
    Started,
    BindFailed,
}

pub struct Server;

impl Server {
    pub fn run(
        listen_addr: ListenAddr,
        gateway_state: Arc<RwLock<manager::GatewayState>>,
        control: Arc<(Mutex<ServerStartup>, Condvar)>,
    ) {
        thread::spawn(move || {
            let &(ref lock, ref cvar) = &*control;
            let sys = actix::System::new("sup-http-gateway");
            let thread_count = match henv::var(HTTP_THREADS_ENVVAR) {
                Ok(val) => match val.parse::<usize>() {
                    Ok(v) => v,
                    Err(_) => HTTP_THREAD_COUNT,
                },
                Err(_) => HTTP_THREAD_COUNT,
            };

            let bind = server::new(move || {
                let app_state = AppState::new(gateway_state.clone());
                App::with_state(app_state).configure(routes)
            }).workers(thread_count)
            .bind(listen_addr.to_string());

            // We need to create this scope on purpose here because if we don't, the lock never
            // releases, and the supervisor will wait forever on cvar. Creating this artifical
            // scope forces the lock to release, and things work as expected.
            {
                let mut started = lock.lock().expect("Control mutex is poisoned");

                *started = match bind {
                    Ok(b) => {
                        b.start();
                        ServerStartup::Started
                    }
                    Err(e) => {
                        error!("HTTP gateway failed to bind: {:?}", e);
                        ServerStartup::BindFailed
                    }
                };
            }

            cvar.notify_one();
            sys.run();
        });
    }
}

struct RedactHTTP;
{'fraction_non_alphanumeric': 0.12436930158449146, 'fraction_numerical': 0.006019297158537665, 'mean_word_length': 2.4277912621359223, 'pattern_counts': {'":': 0, '<': 30, '<?xml version=': 0, '>': 70, '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': '2995375', 'n_tokens_mistral': 3431, 'n_tokens_neox': 3371, 'n_words': 903}