// Run the simulation, generating a "cluster history" and simulated disruptions
Run the simulation, generating a "cluster history" and simulated disruptions
Run the simulation generating a " cluster history " and simulated disruptions
pub fn generate_timeline(client: &Arc<Client>, config: &Config, json: bool) {
    // We generate gaussians on another thread and cache them in a channel
    // to be grabbed when necessary, since gaussian generation is a bit slow
    let rx = start_normal_generator();
    // Threadpool for the client bulks
    let pool = ThreadPool::new(config.threads);
    let mut bulk: Vec<TupleResult> = Vec::with_capacity(config.es.bulk_size);
    let mut rng = thread_rng();
    // Generate a list of disruptions that will be seeded into our timeline
    let disruptions = generate_disruptions(&config, &mut rng);
    // Generate the "regular" and "disrupted" distributions for
    // every (node,metric,query) tupe
    let distributions = generate_distributions(&config, &mut rng);
    // Generate the timeline
    debug!("Generating timeline...");
    let (mut disruption, mut counter) = (None, 0);
    for hour in 0..config.hours {
        debug!("{} -- {}", hour, bulk.len());
        // if we are not currently in a disruption, check to see
        // if there is one this hour
        if counter <= 0 {
            disruption = disruptions.get(&hour);
            counter = disruption.map_or(0, |&(_, x)| x);
        }
        if disruption.is_some() {
            print!(".");
        }
        // iterate through all the (node,query,metric) tuples
        for node in 0..config.nodes {
            for query in 0..config.queries {
                for metric in 0..config.metrics {
                    // Set the disruption flag for this tuple at this hour, based on the
                    // disruption type
                    let d = distributions.get(&(node, query, metric)).unwrap();
                    let is_disrupted = counter > 0 && match disruption {
                        None => false,
                        Some(&(Disruption::Query(ref v), _)) => v.contains(&query),
                        Some(&(Disruption::Metric(ref v), _)) => v.contains(&metric),
                        Some(&(Disruption::Node(ref v), _)) => *v == node
                    };
                    // Grab a gaussian from the normalGenerator thread and use the "regular"
                    // or "disrupted" distributions to find the final value
                    let value = match is_disrupted {
                        true => (rx.recv().unwrap() * d.1.std as f64) + d.1.mean as f64,
                        false => (rx.recv().unwrap() * d.0.std as f64) + d.0.mean as f64
                    };
                    let timestamp = UTC.ymd(2015, 1, 1).and_hms(0, 0, 0) + Duration::hours(hour as i64);
                    bulk.push(TupleResult {
                        node: node,
                        metric: metric,
                        query: query,
                        hour: timestamp.format("%Y-%m-%dT%H:%M:%S").to_string(),
                        value: value,
                        disruption: ::util::disruption_to_usize(&disruption)
                    });
                }
            }
            if bulk.len() >= config.es.bulk_size {
                // Option to dump to a JSON file instead of ES
                if json {
                    let mut file = OpenOptions::new()
                        .read(false)
                        .write(true)
                        .create(true)
                        .append(true)
                        .open("output.json")
                        .unwrap();
                    for b in bulk {
                        let _ = file.write(&json::encode(&b).unwrap().into_bytes());
                        let _ = file.write(b"\n");
                    }
                    bulk = Vec::with_capacity(config.es.bulk_size);
                } else {
                    // Otherwise wait for a free thread and fire off the bulk in the background
                    while ::ACTIVE_THREADS.load(Ordering::SeqCst) >= config.threads {
                        thread::sleep_ms(500);
                    }
                    debug!(".");
                    let client_clone = client.clone();
                    pool.execute(move|| {
                        ::util::send_bulk("http://localhost:9200/data/data/_bulk", &client_clone, bulk);
                    });
                    bulk = Vec::with_capacity(config.es.bulk_size);
                }
            }
        }
        // Decrease the disruption counter.  Disruptions are 2-24 hours long,
        // when counter reaches zero the disruption is over
        if counter > 0 {
            counter -= 1;
        }
    }
    ::GENERATOR_RUNNING.store(false, Ordering::SeqCst);
    ::util::send_bulk("http://localhost:9200/data/data/_bulk", &client, bulk);
}
pub fn generate_timeline ( client : & Arc < Client > , config : & Config , json : bool ) { // We generate gaussians on another thread and cache them in a channel // to be grabbed when necessary, since gaussian generation is a bit slow let rx = start_normal_generator ( ) ; // Threadpool for the client bulks let pool = ThreadPool :: new ( config . threads ) ; let mut bulk : Vec < TupleResult > = Vec :: with_capacity ( config . es . bulk_size ) ; let mut rng = thread_rng ( ) ; // Generate a list of disruptions that will be seeded into our timeline let disruptions = generate_disruptions ( & config , & mut rng ) ; // Generate the "regular" and "disrupted" distributions for // every (node,metric,query) tupe let distributions = generate_distributions ( & config , & mut rng ) ; // Generate the timeline debug ! ( " " ) ; let ( mut disruption , mut counter ) = ( None , 0 ) ; for hour in 0 .. config . hours { debug ! ( " " hour bulk len ( ) ) ; // if we are not currently in a disruption, check to see // if there is one this hour if counter <= 0 { disruption = disruptions . get ( & hour ) ; counter = disruption . map_or ( 0 , | & ( _ , x ) | x ) ; } if disruption . is_some ( ) { print ! ( " " ) ; } // iterate through all the (node,query,metric) tuples for node in 0 .. config . nodes { for query in 0 .. config . queries { for metric in 0 .. config . metrics { // Set the disruption flag for this tuple at this hour, based on the // disruption type let d = distributions . get ( & ( node , query , metric ) ) . unwrap ( ) ; let is_disrupted = counter > 0 && match disruption { None => false , Some ( & ( Disruption :: Query ( ref v ) , _ ) ) => v . contains ( & query ) , Some ( & ( Disruption :: Metric ( ref v ) , _ ) ) => v . contains ( & metric ) , Some ( & ( Disruption :: Node ( ref v ) , _ ) ) => * v == node } ; // Grab a gaussian from the normalGenerator thread and use the "regular" // or "disrupted" distributions to find the final value let value = match is_disrupted { true => ( rx . recv ( ) . unwrap ( ) * d . 1 . std as f64 ) + d . 1 . mean as f64 , false => ( rx . recv ( ) . unwrap ( ) * d . 0 . std as f64 ) + d . 0 . mean as f64 } ; let timestamp = UTC . ymd ( 2015 , 1 , 1 ) . and_hms ( 0 , 0 , 0 ) + Duration :: hours ( hour as i64 ) ; bulk . push ( TupleResult { node : node , metric : metric , query : query , hour : timestamp . format ( " " ) . to_string ( ) , value : value , disruption : :: util :: disruption_to_usize ( & disruption ) } ) ; } } if bulk . len ( ) >= config . es . bulk_size { // Option to dump to a JSON file instead of ES if json { let mut file = OpenOptions :: new ( ) . read ( false ) . write ( true ) . create ( true ) . append ( true ) . open ( " " ) . unwrap ( ) ; for b in bulk { let _ = file . write ( & json :: encode ( & b ) . unwrap ( ) . into_bytes ( ) ) ; let _ = file . write ( b" \n " ) ; } bulk = Vec :: with_capacity ( config . es . bulk_size ) ; } else { // Otherwise wait for a free thread and fire off the bulk in the background while :: ACTIVE_THREADS . load ( Ordering :: SeqCst ) >= config . threads { thread :: sleep_ms ( 500 ) ; } debug ! ( " " ) ; let client_clone = client . clone ( ) ; pool . execute ( move | | { :: util :: send_bulk ( " " , & client_clone , bulk ) ; } ) ; bulk = Vec :: with_capacity ( config . es . bulk_size ) ; } } } // Decrease the disruption counter.  Disruptions are 2-24 hours long, // when counter reaches zero the disruption is over if counter > 0 { counter -= 1 ; } } :: GENERATOR_RUNNING . store ( false , Ordering :: SeqCst ) ; :: util :: send_bulk ( " " , & client , bulk ) ; }
Run the simulation, generating a "cluster history" and simulated disruptions
Run the simulation generating a " cluster history " and simulated disruptions
client
&Arc<Client>
config
&Config
json
bool