pub async fn run(){
  let client = Client::try_default().await.expect("getting default client");
  let namespace = env_var!(required "KUBERNETES_NAMESPACE");
  let deployments: Api<Deployment> = Api::namespaced(client, &namespace);
  let lp = ListParams::default()
    .labels("tibcoems.apimeister.com/scaling=true");
  let mut interval = time::interval(Duration::from_millis(12000));
  interval.tick().await;

  loop{
    interval.tick().await;
    for deployment in deployments.list(&lp).await.unwrap() {
      let deployment_name = ResourceExt::name(&deployment).clone();
      //acquire shared objects
      let mut known_scalings = KNOWN_STATES.lock().unwrap();
      let mut scale_targets = SCALE_TARGETS.lock().unwrap();
      // check if we already know about this deployment
      if known_scalings.contains_key(&deployment_name) {
        let state = known_scalings.get(&deployment_name).unwrap();
        //check replica count and create new state object
        let deployment_state: State;
        let replica_count = deployment.spec.unwrap().replicas.unwrap();
        match (state,replica_count) {
          (State::Active(val), 0) => {
            deployment_state = State::Inactive(StateValue{
              activity_timestamp: get_epoch_seconds(),
              trigger:  val.trigger.clone(),
              deployment: deployment_name.clone(),
              replicas: 0,
              threshold: val.threshold,
              max_scale: val.max_scale,
            });
          },
          (State::Inactive(val), 1) => {
            deployment_state = State::Active(StateValue{
              activity_timestamp: get_epoch_seconds(),
              trigger:  val.trigger.clone(),
              deployment: deployment_name.clone(),
              replicas: 1,
              threshold: val.threshold,
              max_scale: val.max_scale,
            });
          },
          _ => {
            deployment_state = state.clone();
          }
        }
        known_scalings.insert(deployment_name,deployment_state);
      }else{
        info!("Found Deployment: {}", deployment_name);
        //get scale target trigger
        let d_name = deployment_name.clone();
        let mut trigger_map = StateTriggerMap::new();
        let labels = deployment.metadata.labels.unwrap();
        let mut threshold = 100i64;
        let mut max_scale = 10u32;
        for (key,val) in labels {
          if key.starts_with("tibcoems.apimeister.com/queue") {
            //check known queues
            let queue_name = val;
            let all_queues = super::queue::QUEUES.lock().unwrap();
            if all_queues.contains_key(&queue_name) {
              //known queue
              if scale_targets.contains_key(&queue_name) {
                let mut x: Vec<String> =scale_targets.get(&queue_name).unwrap().clone();
                //check if the vector contains the value d_name
                if !x.contains(&d_name) {
                  x.push(d_name.clone());
                }
                info!("add queue scaler queue: {}, deployment: {:?}",queue_name,x);
                scale_targets.insert(queue_name.clone(),x);
              }else{
                info!("add queue scaler queue: {}, deployment: {}",queue_name,d_name);
                scale_targets.insert(queue_name.clone(),vec![d_name.clone()]);
              }
            }else{
              //queue does not exist
              warn!("queue cannot be monitored, because it does not exists: {}",queue_name);
            }
            //add to trigger map
            trigger_map.insert(d_name.clone(),0);
          }else if key.starts_with("tibcoems.apimeister.com/threshold") {
            threshold = match val.parse::<i64>() {
              Ok(result) => result,
              Err(_err) => 100
            };
          }else if key.starts_with("tibcoems.apimeister.com/maxScale") {
            max_scale = match val.parse::<u32>() {
              Ok(result) => result,
              Err(_err) => 10
            };
          }
        }
        //check replica count and create new state object
        let deployment_state: State;
        let replica_count = deployment.spec.unwrap().replicas.unwrap();
        if replica_count == 0 {
          deployment_state = State::Inactive(StateValue{
            activity_timestamp: get_epoch_seconds(),
            trigger:  trigger_map.clone(),
            deployment: deployment_name.clone(),
            replicas: 0,
            threshold,
            max_scale,
          });
        }else{
          deployment_state = State::Active(StateValue{
            activity_timestamp: get_epoch_seconds(),
            trigger: trigger_map.clone(),
            deployment: deployment_name.clone(),
            replicas: 1,
            threshold,
            max_scale,
          });
        }
        if !trigger_map.is_empty() {
          known_scalings.insert(deployment_name, deployment_state);
        }
      }
    }
  }
}
/// watches for k8s Deployments with scaling labels present
watches for k8s Deployments with scaling labels present
watches for k8s Deployments with scaling labels present
pub async fn run(){
  let client = Client::try_default().await.expect("getting default client");
  let namespace = env_var!(required "KUBERNETES_NAMESPACE");
  let deployments: Api<Deployment> = Api::namespaced(client, &namespace);
  let lp = ListParams::default()
    .labels("tibcoems.apimeister.com/scaling=true");
  let mut interval = time::interval(Duration::from_millis(12000));
  interval.tick().await;
  loop{
    interval.tick().await;
    for deployment in deployments.list(&lp).await.unwrap() {
      let deployment_name = ResourceExt::name(&deployment).clone();
      //acquire shared objects
      let mut known_scalings = KNOWN_STATES.lock().unwrap();
      let mut scale_targets = SCALE_TARGETS.lock().unwrap();
      // check if we already know about this deployment
      if known_scalings.contains_key(&deployment_name) {
        let state = known_scalings.get(&deployment_name).unwrap();
        //check replica count and create new state object
        let deployment_state: State;
        let replica_count = deployment.spec.unwrap().replicas.unwrap();
        match (state,replica_count) {
          (State::Active(val), 0) => {
            deployment_state = State::Inactive(StateValue{
              activity_timestamp: get_epoch_seconds(),
              trigger:  val.trigger.clone(),
              deployment: deployment_name.clone(),
              replicas: 0,
              threshold: val.threshold,
              max_scale: val.max_scale,
            });
          },
          (State::Inactive(val), 1) => {
            deployment_state = State::Active(StateValue{
              activity_timestamp: get_epoch_seconds(),
              trigger:  val.trigger.clone(),
              deployment: deployment_name.clone(),
              replicas: 1,
              threshold: val.threshold,
              max_scale: val.max_scale,
            });
          },
          _ => {
            deployment_state = state.clone();
          }
        }
        known_scalings.insert(deployment_name,deployment_state);
      }else{
        info!("Found Deployment: {}", deployment_name);
        //get scale target trigger
        let d_name = deployment_name.clone();
        let mut trigger_map = StateTriggerMap::new();
        let labels = deployment.metadata.labels.unwrap();
        let mut threshold = 100i64;
        let mut max_scale = 10u32;
        for (key,val) in labels {
          if key.starts_with("tibcoems.apimeister.com/queue") {
            //check known queues
            let queue_name = val;
            let all_queues = super::queue::QUEUES.lock().unwrap();
            if all_queues.contains_key(&queue_name) {
              //known queue
              if scale_targets.contains_key(&queue_name) {
                let mut x: Vec<String> =scale_targets.get(&queue_name).unwrap().clone();
                //check if the vector contains the value d_name
                if !x.contains(&d_name) {
                  x.push(d_name.clone());
                }
                info!("add queue scaler queue: {}, deployment: {:?}",queue_name,x);
                scale_targets.insert(queue_name.clone(),x);
              }else{
                info!("add queue scaler queue: {}, deployment: {}",queue_name,d_name);
                scale_targets.insert(queue_name.clone(),vec![d_name.clone()]);
              }
            }else{
              //queue does not exist
              warn!("queue cannot be monitored, because it does not exists: {}",queue_name);
            }
            //add to trigger map
            trigger_map.insert(d_name.clone(),0);
          }else if key.starts_with("tibcoems.apimeister.com/threshold") {
            threshold = match val.parse::<i64>() {
              Ok(result) => result,
              Err(_err) => 100
            };
          }else if key.starts_with("tibcoems.apimeister.com/maxScale") {
            max_scale = match val.parse::<u32>() {
              Ok(result) => result,
              Err(_err) => 10
            };
          }
        }
        //check replica count and create new state object
        let deployment_state: State;
        let replica_count = deployment.spec.unwrap().replicas.unwrap();
        if replica_count == 0 {
          deployment_state = State::Inactive(StateValue{
            activity_timestamp: get_epoch_seconds(),
            trigger:  trigger_map.clone(),
            deployment: deployment_name.clone(),
            replicas: 0,
            threshold,
            max_scale,
          });
        }else{
          deployment_state = State::Active(StateValue{
            activity_timestamp: get_epoch_seconds(),
            trigger: trigger_map.clone(),
            deployment: deployment_name.clone(),
            replicas: 1,
            threshold,
            max_scale,
          });
        }
        if !trigger_map.is_empty() {
          known_scalings.insert(deployment_name, deployment_state);
        }
      }
    }
  }
}
pub async fn run ( ) { let client = Client :: try_default ( ) . await . expect ( " " ) ; let namespace = env_var ! ( required " " ) ; let deployments : Api < Deployment > = Api :: namespaced ( client , & namespace ) ; let lp = ListParams :: default ( ) . labels ( " " ) ; let mut interval = time :: interval ( Duration :: from_millis ( 12000 ) ) ; interval . tick ( ) . await ; loop { interval . tick ( ) . await ; for deployment in deployments . list ( & lp ) . await . unwrap ( ) { let deployment_name = ResourceExt :: name ( & deployment ) . clone ( ) ; //acquire shared objects let mut known_scalings = KNOWN_STATES . lock ( ) . unwrap ( ) ; let mut scale_targets = SCALE_TARGETS . lock ( ) . unwrap ( ) ; // check if we already know about this deployment if known_scalings . contains_key ( & deployment_name ) { let state = known_scalings . get ( & deployment_name ) . unwrap ( ) ; //check replica count and create new state object let deployment_state : State ; let replica_count = deployment . spec . unwrap ( ) . replicas . unwrap ( ) ; match ( state , replica_count ) { ( State :: Active ( val ) , 0 ) => { deployment_state = State :: Inactive ( StateValue { activity_timestamp : get_epoch_seconds ( ) , trigger : val . trigger . clone ( ) , deployment : deployment_name . clone ( ) , replicas : 0 , threshold : val . threshold , max_scale : val . max_scale , } ) ; } , ( State :: Inactive ( val ) , 1 ) => { deployment_state = State :: Active ( StateValue { activity_timestamp : get_epoch_seconds ( ) , trigger : val . trigger . clone ( ) , deployment : deployment_name . clone ( ) , replicas : 1 , threshold : val . threshold , max_scale : val . max_scale , } ) ; } , _ => { deployment_state = state . clone ( ) ; } } known_scalings . insert ( deployment_name , deployment_state ) ; } else { info ! ( " " deployment_name ) ; //get scale target trigger let d_name = deployment_name . clone ( ) ; let mut trigger_map = StateTriggerMap :: new ( ) ; let labels = deployment . metadata . labels . unwrap ( ) ; let mut threshold = 100i64 ; let mut max_scale = 10u32 ; for ( key , val ) in labels { if key . starts_with ( " " ) { //check known queues let queue_name = val ; let all_queues = super :: queue :: QUEUES . lock ( ) . unwrap ( ) ; if all_queues . contains_key ( & queue_name ) { //known queue if scale_targets . contains_key ( & queue_name ) { let mut x : Vec < String > = scale_targets . get ( & queue_name ) . unwrap ( ) . clone ( ) ; //check if the vector contains the value d_name if ! x . contains ( & d_name ) { x . push ( d_name . clone ( ) ) ; } info ! ( " " queue_name x ) ; scale_targets . insert ( queue_name . clone ( ) , x ) ; } else { info ! ( " " queue_name d_name ) ; scale_targets . insert ( queue_name . clone ( ) , vec ! [ d_name clone ( ) ] ) ; } } else { //queue does not exist warn ! ( " " queue_name ) ; } //add to trigger map trigger_map . insert ( d_name . clone ( ) , 0 ) ; } else if key . starts_with ( " " ) { threshold = match val . parse :: < i64 > ( ) { Ok ( result ) => result , Err ( _err ) => 100 } ; } else if key . starts_with ( " " ) { max_scale = match val . parse :: < u32 > ( ) { Ok ( result ) => result , Err ( _err ) => 10 } ; } } //check replica count and create new state object let deployment_state : State ; let replica_count = deployment . spec . unwrap ( ) . replicas . unwrap ( ) ; if replica_count == 0 { deployment_state = State :: Inactive ( StateValue { activity_timestamp : get_epoch_seconds ( ) , trigger : trigger_map . clone ( ) , deployment : deployment_name . clone ( ) , replicas : 0 , threshold , max_scale , } ) ; } else { deployment_state = State :: Active ( StateValue { activity_timestamp : get_epoch_seconds ( ) , trigger : trigger_map . clone ( ) , deployment : deployment_name . clone ( ) , replicas : 1 , threshold , max_scale , } ) ; } if ! trigger_map . is_empty ( ) { known_scalings . insert ( deployment_name , deployment_state ) ; } } } } }
watches for k8s Deployments with scaling labels present
watches for k8s Deployments with scaling labels present