Function elision

pulse-proxy/src/test/integration/multi_node.rs:150–219

()

#[tokio::test]
async fn elision() {
  let (bind_resolver, helper1, helper2) = start_multi_node().await;
  let mut upstream1 = FakeWireUpstream::new("fake_upstream1", bind_resolver.clone()).await;
  let mut stream1 = TcpStream::connect(bind_resolver.local_tcp_addr("inflow:tcp_1"))
    .await
    .unwrap();
  let mut stream2 = TcpStream::connect(bind_resolver.local_tcp_addr("inflow:tcp_2"))
    .await
    .unwrap();

  // 0, 9, and 19 should be reported.
  let current_time = 1;
  let lines = (0 .. 20)
    .map(|i| {
      format!(
        "node_request.count 0 {} source=server-0\n",
        current_time + i
      )
    })
    .collect_vec();

  // To avoid out of order samples we need to wait for every sample to be received before sending
  // the next one.
  for (i, line) in lines.iter().enumerate() {
    if i % 2 == 0 {
      write_all(&mut stream1, &[line]).await;
    } else {
      write_all(&mut stream2, &[line]).await;
    }

    helper1
      .stats_helper()
      .wait_for_counter_eq(
        i as u64 + 1,
        "pulse_proxy:pipeline:messages_routed",
        &labels! {
          "src" => "internode",
          "src_type" => "processor",
          "dest" => "populate_cache",
          "dest_type" => "processor",
        },
      )
      .await;
  }

  assert_eq!(
    parse_carbon_metrics(&[&lines[0], &lines[9], &lines[19]]),
    upstream1.wait_for_num_metrics(3).await
  );

  assert_eq!(
    (current_time + 18).to_string(),
    make_admin_request(
      bind_resolver.local_tcp_addr("admin1"),
      "/last_elided?metric=node_request.count"
    )
    .await
  );