Plugable negotiators - integrate ya-negotiator library

agent/provider/Cargo.toml

@@ -13,7 +13,6 @@ name = "ya-provider"
 path = "src/main.rs"
 
 [dependencies]
-ya-agreement-utils = { version = "0.5.0" }
 ya-manifest-utils = { version = "0.2" }
 ya-client = { version = "0.7", features = ['cli'] }
 ya-client-model = "0.5"
@@ -27,6 +26,12 @@ ya-utils-process = { version = "0.2", features = ['lock'] }
 ya-std-utils = "0.1"
 golem-certificate = "0.1.1"
 
+ya-negotiators = { git = "https://github.com/nieznanysprawiciel/ya-negotiator", rev = "86154fcc8ef2d1142e9f6d2816d7f0cb2ffc560f" }
+ya-agreement-utils = { git = 'https://github.com/nieznanysprawiciel/ya-negotiator', rev = "86154fcc8ef2d1142e9f6d2816d7f0cb2ffc560f" }
+#ya-negotiators = { path = "../../../../experiments/ya-negotiators" }
+#ya-agreement-utils = { path = "../../../../experiments/ya-negotiators/agreement-utils" }
+
+
 actix = { version = "0.13", default-features = false }
 actix-rt = "2.7"
 actix_derive = "0.6"
@@ -42,7 +47,8 @@ dotenv = "0.15.0"
 futures = "0.3"
 futures-util = "0.3.4"
 hex = "0.4"
-humantime = "2.0.0"
+humantime = "2"
+humantime-serde = "1"
 itertools = "0.10"
 lazy_static = "1.4.0"
 libc = "0.2"
@@ -56,6 +62,7 @@ regex = "1.5"
 semver = { version = "0.11", features = ["serde"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
+serde_yaml = "0.9"
 shared_child = "0.3.4"
 signal-hook = "0.3"
 structopt = "0.3.20"

agent/provider/src/execution/registry.rs

@@ -14,7 +14,7 @@ use serde::{Deserialize, Serialize};
 use serde_json::{Map, Value};
 use thiserror::Error;
 
-use ya_agreement_utils::OfferBuilder;
+use crate::typed_props::OfferBuilder;
 
 use super::{exe_unit_work_dir, exeunit_instance::ExeUnitInstance};
 

agent/provider/src/execution/task_runner.rs

@@ -20,7 +20,7 @@ use ya_client::model::activity::provider_event::ProviderEventType;
 use ya_client::model::activity::{ActivityState, ProviderEvent, State, StatePair};
 use ya_std_utils::LogErr;
 use ya_utils_actix::actix_handler::ResultTypeGetter;
-use ya_utils_actix::actix_signal::{Signal, SignalSlot};
+use ya_utils_actix::actix_signal::{SendSignalActor, SignalSlot};
 use ya_utils_actix::{actix_signal_handler, forward_actix_handler};
 use ya_utils_path::SecurePath;
 use ya_utils_process::ExeUnitExitStatus;
@@ -235,13 +235,12 @@ impl TaskRunner {
             .map(|(event, myself)| async move {
                 let _ = match event.event_type {
                     ProviderEventType::CreateActivity { requestor_pub_key } => {
-                        myself
-                            .send(Signal(CreateActivity {
-                                activity_id: event.activity_id,
-                                agreement_id: event.agreement_id,
-                                requestor_pub_key,
-                            }))
-                            .await?
+                        myself.signal(CreateActivity {
+                            activity_id: event.activity_id,
+                            agreement_id: event.agreement_id,
+                            requestor_pub_key,
+                        });
+                        Ok(())
                     }
                     ProviderEventType::DestroyActivity {} => {
                         myself

agent/provider/src/hardware.rs

@@ -13,11 +13,11 @@ use serde::{Deserialize, Serialize};
 use structopt::{clap, StructOpt};
 use tokio::sync::watch;
 
-use ya_agreement_utils::{CpuInfo, InfNodeInfo};
 use ya_utils_path::SwapSave;
 
 use crate::events::Event;
 use crate::startup_config::{FileMonitor, ProviderConfig};
+use crate::typed_props::{CpuInfo, InfNodeInfo};
 
 pub const DEFAULT_PROFILE_NAME: &str = "default";
 pub const CPU_THREADS_RESERVED: i32 = 1;

agent/provider/src/lib.rs

@@ -13,6 +13,7 @@ pub mod rules;
 pub mod signal;
 pub mod startup_config;
 pub mod tasks;
+mod typed_props;
 
 pub use config::globals::GlobalsState;
 pub use startup_config::ReceiverAccount;

agent/provider/src/market/config.rs

@@ -1,6 +1,13 @@
+use serde::{Deserialize, Serialize};
+use std::convert::TryFrom;
+use std::path::{Path, PathBuf};
 use structopt::StructOpt;
 
-use crate::market::negotiator::factory::NegotiatorsConfig;
+use crate::startup_config::{ProviderConfig, DEFAULT_PLUGINS_DIR};
+
+lazy_static::lazy_static! {
+    pub static ref DEFAULT_NEGOTIATORS_PLUGINS_DIR: PathBuf = default_negotiators_plugins();
+}
 
 /// Configuration for ProviderMarket actor.
 #[derive(StructOpt, Clone)]
@@ -11,12 +18,50 @@ pub struct MarketConfig {
     pub negotiation_events_interval: f32,
     #[structopt(long, env, default_value = "10.0")]
     pub agreement_approve_timeout: f32,
-    #[structopt(long, env, default_value = "Composite")]
-    pub negotiator_type: String,
-    #[structopt(flatten)]
-    pub negotiator_config: NegotiatorsConfig,
     #[structopt(skip = "you-forgot-to-set-session-id")]
     pub session_id: String,
     #[structopt(long, env, parse(try_from_str = humantime::parse_duration), default_value = "20s")]
     pub process_market_events_timeout: std::time::Duration,
+    #[structopt(long, env, parse(try_from_str = humantime::parse_duration), default_value = "10s")]
+    pub negotiators_shutdown_timeout: std::time::Duration,
+    /// Relative to Provider DataDir
+    #[structopt(long, env, default_value = "negotiations")]
+    pub negotiators_workdir: String,
+    /// Uses ExeUnit plugins directory by default
+    #[structopt(
+        long,
+        default_value_os = DEFAULT_NEGOTIATORS_PLUGINS_DIR.as_ref(),
+        required = false,
+    )]
+    pub negotiators_plugins: PathBuf,
+    #[structopt(long, env)]
+    pub create_negotiators_config: bool,
+}
+
+/// Agent configuration that will be passed to negotiators.
+#[derive(Clone, Serialize, Deserialize)]
+pub struct AgentNegotiatorsConfig {
+    pub rules_file: PathBuf,
+    pub whitelist_file: PathBuf,
+    pub cert_dir: PathBuf,
+}
+
+fn default_negotiators_plugins() -> PathBuf {
+    PathBuf::from(&*DEFAULT_PLUGINS_DIR)
+        .parent()
+        .map(Path::to_path_buf)
+        .unwrap_or_else(|| "/.local/lib/yagna/plugins/".into())
+}
+
+impl TryFrom<ProviderConfig> for AgentNegotiatorsConfig {
+    type Error = anyhow::Error;
+
+    fn try_from(config: ProviderConfig) -> anyhow::Result<Self> {
+        let cert_dir = config.cert_dir_path()?;
+        Ok(AgentNegotiatorsConfig {
+            rules_file: config.rules_file,
+            whitelist_file: config.domain_whitelist_file,
+            cert_dir,
+        })
+    }
 }

agent/provider/src/market/negotiator.rs

@@ -1,15 +1,5 @@
-mod accept_all;
 pub mod builtin;
 mod common;
-mod component;
-mod composite;
-pub mod factory;
 
-pub use accept_all::AcceptAllNegotiator;
-pub use composite::CompositeNegotiator;
-
-pub use common::{
-    AgreementResponse, AgreementResult, Negotiator, NegotiatorAddr, ProposalResponse,
-};
-
-pub use component::{NegotiationResult, NegotiatorComponent, NegotiatorsPack, ProposalView};
+pub use super::config::AgentNegotiatorsConfig;
+pub use common::AgreementResult;