Merge branch 'master' into scx1332/update_payment7

.github/workflows/integration-test.yml

@@ -78,7 +78,7 @@ jobs:
 
       - name: Build binaries
         run: |
-          cargo build --features static-openssl --target x86_64-unknown-linux-musl -p yagna -p ya-exe-unit -p gftp -p golemsp -p ya-provider -p erc20_processor
+          cargo build --features require-consent,static-openssl --target x86_64-unknown-linux-musl -p yagna -p ya-exe-unit -p gftp -p golemsp -p ya-provider -p erc20_processor
 
       - name: Move target binaries
         run: |
@@ -133,6 +133,7 @@ jobs:
       - name: Check installed binaries
         run: |
           yagna --version
+          yagna consent allow-all
           erc20_processor --version
 
       - name: Run test

.github/workflows/release.yml

@@ -205,9 +205,9 @@ jobs:
       - name: Build macos
         if: matrix.os == 'macos'
         run: |
-          cargo build --release --features static-openssl
+          cargo build --release --features require-consent,static-openssl
           cargo build --bin gftp -p gftp --release
-          cargo build --bin golemsp -p golemsp --release
+          cargo build --bin golemsp --features require-consent -p golemsp --release
           cargo build --bin ya-provider -p ya-provider --release
           cargo build --bin exe-unit -p ya-exe-unit --release --features openssl/vendored
       - name: Build windows
@@ -216,18 +216,18 @@ jobs:
           vcpkg install openssl:x64-windows-static
           vcpkg integrate install
 
-          cargo build --release
+          cargo build --release --features require-consent
           cargo build --bin gftp -p gftp --release
-          cargo build --bin golemsp -p golemsp --release
+          cargo build --bin golemsp --features require-consent -p golemsp --release
           cargo build --bin ya-provider -p ya-provider --release
           cargo build --bin exe-unit -p ya-exe-unit --release
 
       - name: Build linux
         if: matrix.os == 'ubuntu'
         run: |
-          cargo build --release --features static-openssl --target x86_64-unknown-linux-musl
+          cargo build --release --features require-consent,static-openssl --target x86_64-unknown-linux-musl
           (cd core/gftp && cargo build --bin gftp -p gftp --features bin --release --target x86_64-unknown-linux-musl)
-          (cd golem_cli && cargo build --bin golemsp -p golemsp --release --features openssl/vendored --target x86_64-unknown-linux-musl)
+          (cd golem_cli && cargo build --bin golemsp -p golemsp --release --features require-consent,openssl/vendored --target x86_64-unknown-linux-musl)
           (cd agent/provider && cargo build --bin ya-provider -p ya-provider --release --features openssl/vendored --target x86_64-unknown-linux-musl)
           (cd exe-unit && cargo build --bin exe-unit -p ya-exe-unit --release --features openssl/vendored --target x86_64-unknown-linux-musl)
       - name: Pack
@@ -310,7 +310,7 @@ jobs:
             -p golemsp
             -p gftp
             --release
-            --features static-openssl
+            --features require-consent,static-openssl
             --target aarch64-unknown-linux-musl
 
       - name: Pack

.github/workflows/unit-test.yml

@@ -73,4 +73,4 @@ jobs:
         uses: actions-rs/cargo@v1
         with:
           command: test
-          args: --workspace --exclude=["./agent/provider/src/market"] --locked
+          args: --workspace --features require-consent --exclude=["./agent/provider/src/market"] --locked

Cargo.toml

@@ -14,6 +14,7 @@ default = ['erc20-driver', 'gftp/bin']
 dashboard = ['rust-embed', 'mime_guess']
 dummy-driver = ['ya-dummy-driver']
 erc20-driver = ['ya-erc20-driver']
+require-consent = ['ya-utils-consent/require-consent']
 static-openssl = ["openssl/vendored", "openssl-probe"]
 tos = []
 framework-test = [
@@ -56,6 +57,7 @@ ya-service-api-interfaces.workspace = true
 ya-service-api-web.workspace = true
 ya-service-bus = { workspace = true }
 ya-sgx.path = "core/sgx"
+ya-utils-consent.workspace = true
 ya-utils-path.workspace = true
 ya-utils-futures.workspace = true
 ya-utils-process = { workspace = true, features = ["lock"] }
@@ -262,6 +264,7 @@ gftp = {version = "0.4.1", path = "core/gftp"}
 hex = "0.4.3"
 libsqlite3-sys = {version = "0.26.0", features = ["bundled"]}
 openssl = "0.10"
+promptly = "0.3.0"
 rand = "0.8.5"
 regex = "1.10.4"
 strum = {version = "0.24", features = ["derive"]}
@@ -292,6 +295,7 @@ ya-std-utils = { path = "utils/std-utils" }
 ya-diesel-utils.path = "utils/diesel-utils"
 ya-utils-actix.path = "utils/actix_utils"
 ya-core-model = { path = "core/model" }
+ya-utils-consent.path = "utils/consent"
 ya-utils-path.path = "utils/path"
 ya-utils-process.path = "utils/process"
 

core/identity/Cargo.toml

@@ -25,9 +25,9 @@ diesel = { version = "1.4", features = ["sqlite", "r2d2", "chrono"] }
 diesel_migrations = "1.4"
 ethsign = "0.8"
 futures = "0.3"
-hex = { workspace = true }
+hex.workspace = true
 log = "0.4"
-promptly = "0.3.0"
+promptly.workspace = true
 r2d2 = "0.8.8"
 rand = "0.8"
 rpassword = "3.0.2"

core/metrics/Cargo.toml

@@ -14,6 +14,7 @@ ya-core-model = { workspace = true, features = ["identity"] }
 ya-service-api.workspace = true
 ya-service-api-interfaces.workspace = true
 ya-service-bus = {  workspace = true }
+ya-utils-consent = {  workspace = true }
 
 awc = "3"
 actix-web = { version = "4", features = ["openssl"] }

core/metrics/src/pusher.rs

@@ -4,6 +4,7 @@ use lazy_static::lazy_static;
 use percent_encoding::{utf8_percent_encode, AsciiSet, NON_ALPHANUMERIC};
 use tokio::time::{self, Duration, Instant};
 
+use crate::service::export_metrics_for_push;
 use ya_core_model::identity::{self, IdentityInfo};
 use ya_service_api::MetricsCtx;
 use ya_service_bus::typed as bus;
@@ -26,7 +27,7 @@ pub fn spawn(ctx: MetricsCtx) {
             log::warn!("Metrics pusher enabled, but no `push_host_url` provided");
         }
     });
-    log::info!("Metrics pusher started");
+    log::debug!("Metrics pusher started");
 }
 
 pub async fn push_forever(host_url: &str, ctx: &MetricsCtx) {
@@ -54,22 +55,27 @@ pub async fn push_forever(host_url: &str, ctx: &MetricsCtx) {
     let mut push_interval = time::interval_at(start, Duration::from_secs(60));
     let client = Client::builder().timeout(Duration::from_secs(30)).finish();
 
-    log::info!("Starting metrics pusher on address: {push_url}");
+    log::info!(
+        "Metrics will be pushed only if appropriate consent is given, push endpoint: {push_url}"
+    );
     loop {
         push_interval.tick().await;
         push(&client, push_url.clone()).await;
     }
 }
 
 pub async fn push(client: &Client, push_url: String) {
-    let metrics = crate::service::export_metrics().await;
+    let metrics = export_metrics_for_push().await;
+    if metrics.is_empty() {
+        return;
+    }
     let res = client
         .put(push_url.as_str())
         .send_body(metrics.clone())
         .await;
     match res {
         Ok(r) if r.status().is_success() => {
-            log::trace!("Metrics pushed: {}", r.status())
+            log::debug!("Metrics pushed: {}", r.status())
         }
         Ok(r) if r.status().is_server_error() => {
             log::debug!("Metrics server error: {:#?}", r);

core/metrics/src/service.rs

@@ -1,3 +1,4 @@
+use actix_web::web::Path;
 use futures::lock::Mutex;
 use lazy_static::lazy_static;
 use std::collections::HashMap;
@@ -7,6 +8,7 @@ use url::Url;
 
 use ya_service_api::{CliCtx, MetricsCtx};
 use ya_service_api_interfaces::Provider;
+use ya_utils_consent::ConsentScope;
 
 use crate::metrics::Metrics;
 
@@ -72,6 +74,15 @@ lazy_static! {
     static ref METRICS: Arc<Mutex<Metrics>> = Metrics::new();
 }
 
+pub async fn export_metrics_filtered_web(typ: Path<String>) -> String {
+    let allowed_prefixes = typ.split(',').collect::<Vec<_>>();
+    log::info!("Allowed prefixes: {:?}", allowed_prefixes);
+    let filter = MetricsFilter {
+        allowed_prefixes: &allowed_prefixes,
+    };
+    export_metrics_filtered(Some(filter)).await
+}
+
 impl MetricsService {
     pub async fn gsb<C: Provider<Self, CliCtx>>(context: &C) -> anyhow::Result<()> {
         // This should initialize Metrics. We need to do this before all other services will start.
@@ -89,35 +100,86 @@ impl MetricsService {
     pub fn rest<C: Provider<Self, ()>>(_ctx: &C) -> actix_web::Scope {
         actix_web::Scope::new("metrics-api/v1")
             // TODO:: add wrapper injecting Bearer to avoid hack in auth middleware
-            .route("/expose", actix_web::web::get().to(export_metrics))
+            .route("/expose", actix_web::web::get().to(export_metrics_local))
             .route("/sorted", actix_web::web::get().to(export_metrics_sorted))
+            .route(
+                "/filtered/{typ}",
+                actix_web::web::get().to(export_metrics_filtered_web),
+            )
+            .route(
+                "/filtered",
+                actix_web::web::get().to(export_metrics_for_push),
+            )
     }
 }
+
+pub(crate) struct MetricsFilter<'a> {
+    pub allowed_prefixes: &'a [&'a str],
+}
+
 //algorith is returning metrics in random order, which is fine for prometheus, but not for human checking metrics
-pub fn sort_metrics_txt(metrics: &str) -> String {
+pub fn sort_metrics_txt(metrics: &str, filter: Option<MetricsFilter<'_>>) -> String {
     let Some(first_line_idx) = metrics.find('\n') else {
         return metrics.to_string();
     };
     let (first_line, metrics_content) = metrics.split_at(first_line_idx);
 
-    let mut entries = metrics_content
+    let entries = metrics_content
         .split("\n\n") //splitting by double new line to get separate metrics
         .map(|s| {
             let trimmed = s.trim();
             let mut lines = trimmed.split('\n').collect::<Vec<_>>();
             lines.sort(); //sort by properties
-            lines.join("\n")
+            (lines.get(1).unwrap_or(&"").to_string(), lines.join("\n"))
         })
-        .collect::<Vec<String>>();
-    entries.sort(); //sort by metric name
+        .collect::<Vec<(String, String)>>();
+
+    let mut final_entries = if let Some(filter) = filter {
+        let mut final_entries = Vec::with_capacity(entries.len());
+        for entry in entries {
+            for prefix in filter.allowed_prefixes {
+                if entry.0.starts_with(prefix) {
+                    log::info!("Adding entry: {}", entry.0);
+                    final_entries.push(entry.1);
+                    break;
+                }
+            }
+        }
+        final_entries
+    } else {
+        entries.into_iter().map(|(_, s)| s).collect()
+    };
 
-    first_line.to_string() + "\n" + entries.join("\n\n").as_str()
+    final_entries.sort();
+
+    first_line.to_string() + "\n" + final_entries.join("\n\n").as_str() + "\n"
+}
+
+pub async fn export_metrics_filtered(metrics_filter: Option<MetricsFilter<'_>>) -> String {
+    sort_metrics_txt(&METRICS.lock().await.export(), metrics_filter)
 }
 
 async fn export_metrics_sorted() -> String {
-    sort_metrics_txt(&METRICS.lock().await.export())
+    sort_metrics_txt(&METRICS.lock().await.export(), None)
+}
+
+pub async fn export_metrics_for_push() -> String {
+    //if consent is not set assume we are not allowed to push metrics
+    let stats_consent = ya_utils_consent::have_consent_cached(ConsentScope::Stats)
+        .consent
+        .unwrap_or(false);
+    let filter = if stats_consent {
+        log::info!("Pushing all metrics, because stats consent is given");
+        None
+    } else {
+        // !internal_consent && !external_consent
+        log::info!("Not pushing metrics, because stats consent is not given");
+        return "".to_string();
+    };
+
+    export_metrics_filtered(filter).await
 }
 
-pub async fn export_metrics() -> String {
-    METRICS.lock().await.export()
+pub async fn export_metrics_local() -> String {
+    export_metrics_sorted().await
 }

core/serv/src/main.rs

@@ -53,6 +53,9 @@ use autocomplete::CompleteCommand;
 
 use ya_activity::TrackerRef;
 use ya_service_api_web::middleware::cors::AppKeyCors;
+use ya_utils_consent::{
+    consent_check_before_startup, set_consent_path_in_yagna_dir, ConsentService,
+};
 
 lazy_static::lazy_static! {
     static ref DEFAULT_DATA_DIR: String = DataDir::new(clap::crate_name!()).to_string();
@@ -261,6 +264,8 @@ enum Services {
     Activity(ActivityService),
     #[enable(gsb, rest, cli)]
     Payment(PaymentService),
+    #[enable(cli)]
+    Consent(ConsentService),
     #[enable(gsb)]
     SgxDriver(SgxService),
     #[enable(gsb, rest)]
@@ -475,6 +480,7 @@ impl ServiceCommand {
         if !ctx.accept_terms {
             prompt_terms()?;
         }
+
         match self {
             Self::Run(ServiceCommandOpts {
                 api_url,
@@ -541,6 +547,9 @@ impl ServiceCommand {
 
                 let _lock = ProcLock::new(app_name, &ctx.data_dir)?.lock(std::process::id())?;
 
+                //before running yagna check consents
+                consent_check_before_startup(false)?;
+
                 ya_sb_router::bind_gsb_router(ctx.gsb_url.clone())
                     .await
                     .context("binding service bus router")?;
@@ -761,6 +770,7 @@ async fn main() -> Result<()> {
 
     std::env::set_var(GSB_URL_ENV_VAR, args.gsb_url.as_str()); // FIXME
 
+    set_consent_path_in_yagna_dir()?;
     match args.run_command().await {
         Ok(()) => Ok(()),
         Err(err) => {

extra/payments/multi_test/payment_test.py

@@ -167,6 +167,10 @@ def process_erc20():
     balance = get_balance()
     if balance[public_addrs[0]]["tokenDecimal"] != "0":
         raise Exception("Test failed early because of wrong initial balance")
+
+    # give consent before running yagna service
+    run_command(f"{yagna} consent allow-all")
+
     pr = subprocess.Popen([yagna, "service", "run"])
     time.sleep(10)