feat: implement policy merge in dataflow dynamics

.rubocop_todo.yml

@@ -356,7 +356,6 @@ Lint/MissingSuper:
     - 'lib/syskit/robot/master_device_instance.rb'
     - 'lib/syskit/robot/slave_device_instance.rb'
     - 'lib/syskit/telemetry/agent/server.rb'
-    - 'lib/syskit/telemetry/ui/name_service.rb'
     - 'lib/syskit/test/stub_network.rb'
 
 # Offense count: 2

Rakefile

@@ -28,26 +28,46 @@ def minitest_set_options(test_task, name)
     test_task.options = "#{TESTOPTS} #{minitest_args} -- --simplecov-name=#{name}"
 end
 
-Rake::TestTask.new("test:core") do |t|
+def core(early_deploy: false)
+    s = ":no-early-deploy"
+    if early_deploy
+        s = ":early-deploy"
+        early_deploy_setup = ["test/features/early_deploy.rb"]
+    end
+
+    Rake::TestTask.new("test:core#{s}") do |t|
+        t.libs << "."
+        t.libs << "lib"
+        minitest_set_options(t, "core")
+        test_files = FileList["test/**/test_*.rb", *early_deploy_setup]
+        test_files = test_files
+                     .exclude("test/ros/**/*.rb")
+                     .exclude("test/gui/**/*.rb")
+                     .exclude("test/live/**/*.rb")
+                     .exclude("test/telemetry/**/*.rb")
+        t.test_files = test_files
+        t.warning = false
+    end
+end
+
+Rake::TestTask.new("test:telemetry") do |t|
     t.libs << "."
     t.libs << "lib"
-    minitest_set_options(t, "core")
-    test_files = FileList["test/**/test_*.rb"]
-    test_files = test_files
-                 .exclude("test/ros/**/*.rb")
-                 .exclude("test/gui/**/*.rb")
-                 .exclude("test/live/**/*.rb")
-    t.test_files = test_files
+    minitest_set_options(t, "telemetry")
+    t.test_files = FileList["test/telemetry/**/test_*.rb"]
     t.warning = false
 end
 
+desc "Run separate tests that require a live syskit instance"
 task "test:live" do
     tests = Dir.enum_for(:glob, "test/live/test_*.rb").to_a
     unless system(File.join("test", "live", "run"), *tests)
         $stderr.puts "live tests failed"
         exit 1
     end
 end
+
+desc "run gui-only tests"
 Rake::TestTask.new("test:gui") do |t|
     t.libs << "."
     t.libs << "lib"
@@ -57,7 +77,13 @@ Rake::TestTask.new("test:gui") do |t|
     t.warning = false
 end
 
-task "test" => ["test:gui", "test:core", "test:live"]
+core early_deploy: true
+core
+desc "Run core library tests, excluding GUI and live tests"
+task "test:core" => ["test:core:no-early-deploy", "test:core:early-deploy"]
+
+desc "Run all tests"
+task "test" => ["test:gui", "test:core", "test:live", "test:telemetry"]
 
 task "rubocop" do
     raise "rubocop failed" unless system(ENV["RUBOCOP_CMD"] || "rubocop")

lib/syskit.rb

@@ -42,7 +42,7 @@ module ProcessManagers
 require "syskit/roby_app/log_transfer_server"
 require "syskit/process_managers/process_base"
 require "syskit/process_managers/status"
-require "syskit/process_managers/remote/server/log_upload_state"
+require "syskit/roby_app/log_transfer_server/log_upload_state"
 require "syskit/process_managers/remote/protocol"
 require "syskit/process_managers/remote/loader"
 require "syskit/process_managers/remote/manager"
@@ -127,6 +127,7 @@ module ProcessManagers
 require "syskit/actual_data_flow_graph"
 require "syskit/data_flow"
 require "syskit/connection_graphs"
+require "syskit/network_generation_exception_helpers"
 require "syskit/exceptions"
 require "syskit/network_generation"
 require "syskit/runtime"

lib/syskit/base.rb

@@ -2,6 +2,7 @@
 
 require "logger"
 require "utilrb/logger"
+require "syskit/network_generation_exception_helpers"
 require "syskit/exceptions"
 require "facets/string/snakecase"
 

lib/syskit/cli/log_runtime_archive.rb

@@ -2,6 +2,7 @@
 
 require "archive/tar/minitar"
 require "sys/filesystem"
+require "syskit/roby_app/log_transfer_server/ftp_upload"
 
 module Syskit
     module CLI
@@ -17,16 +18,40 @@ class CompressionFailed < RuntimeError; end
         class LogRuntimeArchive
             DEFAULT_MAX_ARCHIVE_SIZE = 10_000_000_000 # 10G
 
+            FTPParameters = Struct.new(:host, :port, :certificate, :user, :password,
+                                       :implicit_ftps, :max_upload_rate,
+                                       keyword_init: true)
+
+            # Initializes the LogRuntimeArchive
+            #
+            # @param [Pathname] root_dir the logs directory
+            # @param [Pathname] target_dir the path to store the file in the archive,
+            #   should be nil in transfer mode, as the logs will be transferred directly
+            #   to the ftp server @see process_root_folder_transfer
+            # @param [Logger] logger the log structure
             def initialize(
-                root_dir, target_dir,
-                logger: LogRuntimeArchive.null_logger,
-                max_archive_size: DEFAULT_MAX_ARCHIVE_SIZE
+                root_dir, target_dir: nil,
+                logger: LogRuntimeArchive.null_logger
             )
                 @last_archive_index = {}
                 @logger = logger
                 @root_dir = root_dir
                 @target_dir = target_dir
-                @max_archive_size = max_archive_size
+            end
+
+            # Iterate over all datasets in a Roby log root folder and transfer them
+            # through FTP server
+            #
+            # @param [Params] server_params the FTP server parameters
+            # @return [Array<TransferDatasetResult>]
+            def process_root_folder_transfer(server_params)
+                candidates = self.class.find_all_dataset_folders(@root_dir)
+                running = candidates.last
+                candidates.map do |child|
+                    process_dataset_transfer(
+                        child, server_params, @root_dir, full: child != running
+                    )
+                end
             end
 
             # Iterate over all datasets in a Roby log root folder and archive them
@@ -37,11 +62,14 @@ def initialize(
             # @param [Pathname] root_dir the log root folder
             # @param [Pathname] target_dir the folder in which to save the
             #   archived datasets
-            def process_root_folder
+            # @param [Integer] max_archive_size the max size of the archive
+            def process_root_folder(max_archive_size: DEFAULT_MAX_ARCHIVE_SIZE)
                 candidates = self.class.find_all_dataset_folders(@root_dir)
                 running = candidates.last
                 candidates.each do |child|
-                    process_dataset(child, full: child != running)
+                    process_dataset(
+                        child, max_archive_size: max_archive_size, full: child != running
+                    )
                 end
             end
 
@@ -54,48 +82,55 @@ def process_root_folder
             #   bytes, at which the archiver starts deleting the oldest log files
             # @param [integer] free_space_delete_until: post-deletion free space in bytes,
             #   at which the archiver stops deleting the oldest log files
-            def ensure_free_space(free_space_low_limit, free_space_delete_until)
+            #
+            # @return [Boolean] true if successfully ensured free space, meaning there is
+            # the required free space, false if deleting the files in this directory was
+            # not enough to free up the required space
+            def ensure_free_space(
+                free_space_low_limit, free_space_delete_until, directory: @target_dir
+            )
                 if free_space_low_limit > free_space_delete_until
                     raise ArgumentError,
                           "cannot erase files: freed limit is smaller than " \
                           "low limit space."
                 end
 
-                stat = Sys::Filesystem.stat(@target_dir)
+                stat = Sys::Filesystem.stat(directory)
                 available_space = stat.bytes_available
 
-                return if available_space > free_space_low_limit
+                return true if available_space > free_space_low_limit
 
                 until available_space >= free_space_delete_until
-                    files = @target_dir.each_child.select(&:file?)
+                    files = directory.each_child.select(&:file?)
                     if files.empty?
                         Roby.warn "Cannot erase files: the folder is empty but the " \
                                   "available space is smaller than the threshold."
-                        break
+                        return false
                     end
 
-                    removed_file = files.min
+                    removed_file = files.min_by(&:mtime)
                     size_removed_file = removed_file.size
                     removed_file.unlink
                     available_space += size_removed_file
                 end
+                true
             end
 
-            def process_dataset(child, full:)
+            def process_dataset(child, full:, max_archive_size: DEFAULT_MAX_ARCHIVE_SIZE)
                 use_existing = true
                 loop do
                     open_archive_for(
                         child.basename.to_s, use_existing: use_existing
                     ) do |io|
-                        if io.tell > @max_archive_size
+                        if io.tell > max_archive_size
                             use_existing = false
                             break
                         end
 
                         dataset_complete = self.class.archive_dataset(
                             io, child,
                             logger: @logger, full: full,
-                            max_size: @max_archive_size
+                            max_size: max_archive_size
                         )
                         return if dataset_complete
                     end
@@ -104,6 +139,99 @@ def process_dataset(child, full:)
                 end
             end
 
+            def process_dataset_transfer(child, server, root, full:)
+                self.class.transfer_dataset(
+                    child, server, root, full: full, logger: @logger
+                )
+            end
+
+            TransferDatasetResult = Struct.new(
+                :complete, :transfer_results, keyword_init: true
+            ) do
+                def success?
+                    transfer_results.all?(&:success?)
+                end
+
+                def failures
+                    transfer_results.find_all { !_1.success? }
+                end
+            end
+
+            # Transfer the given dataset
+            def self.transfer_dataset(
+                dataset_path, server, root,
+                full:, logger: null_logger
+            )
+                logger.info(
+                    "Transfering dataset #{dataset_path} in " \
+                    "#{full ? 'full' : 'partial'} mode"
+                )
+                candidates = each_file_from_path(dataset_path).to_a
+
+                complete, candidates =
+                    if full
+                        archive_filter_candidates_full(candidates)
+                    else
+                        archive_filter_candidates_partial(candidates)
+                    end
+
+                transfer_results = candidates.map do |child_path|
+                    result = transfer_file(child_path, server, root)
+                    child_path.unlink if result.success?
+
+                    result
+                end
+
+                result = TransferDatasetResult.new(
+                    complete: complete, transfer_results: transfer_results
+                )
+                log_transfer_results(dataset_path, result, logger: logger)
+            end
+
+            # Logs the transfer dataset results
+            #
+            # @param [String] the dataset path
+            # @param [TransferDatasetResult] the transfer dataset result
+            # @param [Logger] optional logger, if unfilled will use null logger
+            #
+            # @result [TransferDatasetResult] the received transfer dataset result
+            def self.log_transfer_results(dataset_path, result, logger: null_logger)
+                failed_results = result[:transfer_results].reject(&:success)
+
+                if failed_results.empty?
+                    logger.info(
+                        "Transfering of " \
+                        "#{result[:complete] ? 'complete' : 'incomplete'} " \
+                        "#{dataset_path} finished"
+                    )
+                else
+                    failed_results.each do |failed_result|
+                        failed_message =
+                            if failed_result.message
+                                "with message : #{failed_result.message}"
+                            end
+                        logger.info(
+                            "Failed on file #{failed_result.file} #{failed_message}"
+                        )
+                    end
+                end
+
+                result
+            end
+
+            # Transfer a file to the central log server via FTP
+            #
+            # @return [LogUploadState:Result]
+            def self.transfer_file(file, server, root)
+                ftp = RobyApp::LogTransferServer::FTPUpload.new(
+                    server.host, server.port, server.certificate, server.user,
+                    server.password, file,
+                    max_upload_rate: server.max_upload_rate || Float::INFINITY,
+                    implicit_ftps: server.implicit_ftps
+                )
+                ftp.open_and_transfer(root: root)
+            end
+
             # Create or open an archive
             #
             # The method will find an archive to open or create, do it and
@@ -158,7 +286,7 @@ def self.find_all_dataset_folders(root_dir)
                     child if (child / "info.yml").file?
                 end
 
-                candidates.compact.sort_by { _1.basename.to_s }
+                candidates.compact.sort_by { |a| a.basename.to_s }
             end
 
             # Safely add an entry into an archive, compressing it with zstd