# BrokerManager.java

    /*
    ** Module   : BrokerManager.java
    ** Abstract : server side broker manager implementation.
    **
    ** Copyright (c) 2014-2024, Golden Code Development Corporation.

    **
    ** -#- -I- --Date-- ---------------------------------Description----------------------------------
    ** 001 MAG 20140707 Implements remote launcher (broker).
    ** 002 SBI 20171022 Added dedicated hosts map and changed broker client schedule method to launch
    **                  remote clients on dedicated hosts, enumerated assigned hosts and added
    **                  getOrderedNumber to get the host order number.
    ** 003 SBI 20200317 Added the hosts manager to separate the business logic.
    ** 004 SBI 20230505 Added new spawn method to allocate network ports before the client process is
    **                  spawned, added updateWebClientOptions.
    ** 005 GBB 20230512 Logging methods replaced by CentralLogger/ConversionStatus.
    ** 006 SBI 20230615 Fixed the incorrect error code that hides the spawner error code.
    ** 007 GBB 20230825 WebHandler renamed to WebDriverHandler due to collision in names.
    ** 008 GBB 20240709 CLIENT_IP constant moved here from WebDriverHandler.
    **                  WebClientConfig renamed to WebAllocatedResources.
    **                  Hard-coded config names replaced by ConfigItem constants.
    */
    /*
    ** This program is free software: you can redistribute it and/or modify
    ** it under the terms of the GNU Affero General Public License as
    ** published by the Free Software Foundation, either version 3 of the
    ** License, or (at your option) any later version.
    **
    ** This program is distributed in the hope that it will be useful,
    ** but WITHOUT ANY WARRANTY; without even the implied warranty of
    ** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    ** GNU Affero General Public License for more details.
    **
    ** You may find a copy of the GNU Affero GPL version 3 at the following
    ** location: https://www.gnu.org/licenses/agpl-3.0.en.html
    **
    ** Additional terms under GNU Affero GPL version 3 section 7:
    **
    **   Under Section 7 of the GNU Affero GPL version 3, the following additional
    **   terms apply to the works covered under the License.  These additional terms
    **   are non-permissive additional terms allowed under Section 7 of the GNU
    **   Affero GPL version 3 and may not be removed by you.
    **
    **   0. Attribution Requirement.
    **
    **     You must preserve all legal notices or author attributions in the covered
    **     work or Appropriate Legal Notices displayed by works containing the covered
    **     work.  You may not remove from the covered work any author or developer
    **     credit already included within the covered work.
    **
    **   1. No License To Use Trademarks.
    **
    **     This license does not grant any license or rights to use the trademarks
    **     Golden Code, FWD, any Golden Code or FWD logo, or any other trademarks
    **     of Golden Code Development Corporation. You are not authorized to use the
    **     name Golden Code, FWD, or the names of any author or contributor, for
    **     publicity purposes without written authorization.
    **
    **   2. No Misrepresentation of Affiliation.
    **
    **     You may not represent yourself as Golden Code Development Corporation or FWD.
    **
    **     You may not represent yourself for publicity purposes as associated with
    **     Golden Code Development Corporation, FWD, or any author or contributor to
    **     the covered work, without written authorization.
    **
    **   3. No Misrepresentation of Source or Origin.
    **
    **     You may not represent the covered work as solely your work.  All modified
    **     versions of the covered work must be marked in a reasonable way to make it
    **     clear that the modified work is not originating from Golden Code Development
    **     Corporation or FWD.  All modified versions must contain the notices of
    **     attribution required in this license.
    */

    package com.goldencode.p2j.main;

    import java.net.*;
    import java.util.*;
    import java.util.concurrent.*;
    import java.util.logging.*;

    import com.goldencode.p2j.directory.*;
    import com.goldencode.p2j.net.*;
    import com.goldencode.p2j.security.SecurityManager;
    import com.goldencode.p2j.util.*;
    import com.goldencode.p2j.util.logging.*;
    import static com.goldencode.p2j.main.SpawnError.*;

    /**
     * This class has been designed to manage remote launchers (brokers).
     */
    public class BrokerManager
    {
       /** The client address parameter key */
       static final String CLIENT_IP = "CLIENT_IP";

       /** Logger. */
       private static final CentralLogger LOG = CentralLogger.get(BrokerManager.class.getName());

       /** Brokers registry */
       private static Map<String, Broker> activeBrokers = new ConcurrentHashMap<>();

       /** Map of users to dedicated hosts */
       private static Map<String, List<String>> dedicatedHosts = new LinkedHashMap<String, List<String>>();

       /** Map of users to hosts */
       private static Map<String, List<String>> allHosts = new LinkedHashMap<String, List<String>>();

       /** Manages registered brokers agents */
       private static HostsManager hostsManager;

       /** Session listener */
       private static SessionListener sessionListener = new SessionListener()
       {
          @Override
          public void initialize(Session session)
          {
             // nothings
          }

          @Override
          public void terminate(Session session)
          {
             // unregister broker on session close.
             removeBroker(session);
          }
       };

       /**
        * Read brokers configuration from directory. The method is called on server startup via
        * a registered server hook.
        */
       public static void initialize()
       {
          synchronized (activeBrokers)
          {
             DirectoryService ds = DirectoryService.getInstance();

             hostsManager = HostsManager.getInstance();

             // find brokers path
             String path = Utils.findDirectoryNodePath(ds, "brokers", "container", false);
             if (path == null)
             {
                return;
             }

             // get brokers
             String[] brokers = ds.enumerateNodes(path);
             if (brokers == null)
             {
                return;
             }

             for (String broker : brokers)
             {
                // broker path
                StringBuilder brokerPath = new StringBuilder(path);

                if (!path.endsWith("/"))
                {
                   brokerPath.append("/");
                }

                brokerPath.append(broker);

                // get the list of accounts if exists and store.
                List<String> accounts = getBrokerAccounts(ds, brokerPath.toString());

                activeBrokers.put(broker, new Broker(accounts));
             }
          }
       }

       /**
        * Called on server shutdown from a server hook.
        */
       public static void cleanup()
       {
          synchronized (activeBrokers)
          {
             for (Map.Entry<String, Broker> entry : activeBrokers.entrySet())
             {
                Broker broker = entry.getValue();
                broker.shutdown();
             }
          }
       }

       /**
        * Register a remote launcher (broker). This is the implementation of:
        * {@link BrokerServerServices#registerBroker}
        *
        * @param    host
        *           The host address of the remote launcher
        * @param    user
        *           P2J or OS user account associated with this registering broker
        * @param    dedicatedMode
        *           True value indicates that this registering broker client will spawn clients
        *           accessed only from its local host, otherwise spawned clients can be accessed
        *           from network.
        *
        * @return  The assigned UUID which make broker unique.
        *
        * @throws  UnknownHostException
        *          The host is not valid.
        */
       public static String registerBroker(String host, String user, boolean dedicatedMode)
       throws UnknownHostException
       {
          synchronized (activeBrokers)
          {
             // check if brokers are defined for this P2J server.
             if (activeBrokers.isEmpty())
             {
                throw new RuntimeException("No brokers are defined for this P2J server.");
             }

             SecurityManager sm = SecurityManager.getInstance();

             BrokerParameters params = new BrokerParameters();

             // account id
             params.setUserId(sm.getUserId());
             if (!sm.isProcessDefined(params.getUserId()))
             {
                throw new RuntimeException("Only process accounts are allowed.");
             }

             // broker name if any
             params.setBrokerName(sm.getBrokerForProcess(params.getUserId()));
             if (params.getBrokerName() == null)
             {
                throw new RuntimeException("No broker is defined for this process account.");
             }

             // always add to all hosts first
             addUserHostWithCheck(user, host, allHosts);
             //
             if (dedicatedMode)
             {
                addUserHostWithCheck(user, host, dedicatedHosts);
             }

             // register
             Broker broker = activeBrokers.get(params.getBrokerName());

             if (broker == null)
             {
                throw new RuntimeException("No broker is defined for this process account.");
             }

             if (!broker.hasAccount(user))
             {
                broker.addAccount(user);
             }

             params.setAssociatedUser(user);

             broker.addBroker(params);

             hostsManager.addHost(host, WebClientsManager.getInstance().createResourceQueue(host));

             // add listener to broker session
             Session session = SessionManager.get().getSession();
             session.addSessionListener(sessionListener);
             params.setSession(session);

             // get broker export services
             params.setRemote((BrokerClientServices) RemoteObject.obtainNetworkInstance(
                                                             BrokerClientServices.class,
                                                             session));

             // log informations about registered broker
             LOG.logp(Level.INFO,
                      "BrokerManager.registerBroker()",
                      "",
                      CentralLogger.generate("Broker user=%s uuid=%s registered.",
                                             params.getUserId(),
                                             params.getUuid()));
             return params.getUuid();
          }
       }

       /**
        * Open a communication route to broker.
        *
        * @param   uuid
        *          Broker UUID.
        */
       public static void start(String uuid)
       {
          Session session = SessionManager.get().getSession();

          BrokerParameters broker = findBroker(uuid);
          if (broker != null && session.equals(broker.getSession()))
          {
             broker.open();
          }
       }

       /**
        * Spawn a process remotely via a broker. For generic brokers the brokers from list are used
        * one by one based on system loading reported by broker until the spawn return success or all
        * brokers are used. If all brokers are used without success the last result is returned.
        * <p>
        * For brokers having the user name defined in the list of accounts a broker is selected from
        * list based on system loading reported by broker and this broker is used to launch the remote
        * client. On error no other brokers are selected.
        *
        * @param    dedicatedUser
        *           The dedicated user associated with the host on which the spawned client
        *           will be run
        * @param    brokers
        *           List of available brokers.
        * @param    args
        *           Process parameters.
        *
        * @return  The result of remote spawn.
        */
       public static BrokerSpawnResult spawn(String dedicatedUser,
                                             Brokers brokers,
                                             BrokerSpawnParameters args)
       {
          return spawn(dedicatedUser, brokers, args, null, null, null);
       }

       /**
        * Spawn a process remotely via a broker. For generic brokers the brokers from list are used
        * one by one based on system loading reported by broker until the spawn return success or all
        * brokers are used. If all brokers are used without success the last result is returned.
        * <p>
        * For brokers having the user name defined in the list of accounts a broker is selected from
        * list based on system loading reported by broker and this broker is used to launch the remote
        * client. On error no other brokers are selected.
        *
        * @param    dedicatedUser
        *           The dedicated user associated with the host on which the spawned client
        *           will be run
        * @param    brokers
        *           List of available brokers.
        * @param    args
        *           Process parameters.
        * @param    allocator
        *           The web client allocator
        * @param    requestParameters
        *           The given requests parameters
        * @param    uuid
        *           The uuid of the spawned client
        *
        * @return  The result of remote spawn.
        */
       public static BrokerSpawnResult spawn(String dedicatedUser,
                                             Brokers brokers,
                                             BrokerSpawnParameters args,
                                             WebClientAllocator allocator,
                                             String[] requestParameters,
                                             String uuid)
       {
          // list of available brokers
          List<BrokerParameters> brokersList = brokers.getBrokers();

          // spawn results
          BrokerSpawnResult result = null;

          String clientAddress = args.getEnvironment().get(CLIENT_IP);

          // check if generic brokers
          if (brokers.isGeneric())
          {
             do
             {
                BrokerParameters broker = null;

                try
                {
                   // spawn
                   broker = schedule(brokersList, dedicatedUser, clientAddress);
                   boolean spawn = true;
                   if (allocator != null)
                   {
                      WebAllocatedResources clientConfig = allocator.allocateClient(requestParameters,
                                                                                    broker.getRemote().getDedicatedHost(),
                                                                                    uuid,
                                                                                    dedicatedUser);
                      spawn = (clientConfig != null);
                      updateWebClientOptions(args, clientConfig);
                   }
                   if (spawn)
                   {
                      result = broker.getRemote().spawn(args);
                   }
                }
                catch (Throwable t)
                {
                   LOG.logp(Level.SEVERE,
                            "BrokerManager.schedule()",
                            "",
                            "Remote broker call exception.",
                            t);
                }

                // on success exit loop
                if (result != null && result.getExitCode() == 0)
                {
                   break;
                }

                // on error remove the broker from list and try with the next broker.
                brokersList.remove(broker);
             }
             while (!brokersList.isEmpty());
          }
          else
          {
             try
             {
                // select a broker from list and spawn.
                BrokerParameters broker = schedule(brokersList, dedicatedUser, clientAddress);
                boolean spawn = true;
                if (allocator != null)
                {
                   WebAllocatedResources clientConfig = allocator.allocateClient(requestParameters,
                                                                                 broker.getRemote().getDedicatedHost(),
                                                                                 uuid,
                                                                                 dedicatedUser);
                   spawn = (clientConfig != null);
                   updateWebClientOptions(args, clientConfig);
                }
                if (spawn)
                {
                   result = broker.getRemote().spawn(args);
                }
             }
             catch (Throwable t)
             {
                LOG.logp(Level.SEVERE,
                         "BrokerManager.spawn()",
                         "",
                         "Remote broker call exception.",
                         t);
             }
          }

          // check if any result
          if (result == null)
          {
             result = new BrokerSpawnResult();
             result.setExitCode(CAN_NOT_SPAWN_REMOTE_ERR_CODE.getCode());
             result.setException(new NullPointerException("Cannot spawn remote client."));
          }

          return result;
       }

       /**
        * Get a list of brokers for the given P2J or OS user account.
        *
        * @param   user
        *          P2J or OS user account
        *
        * @return  List of brokers or an empty list if no brokers.
        */
       public static Brokers getBrokers(String user)
       {
          synchronized (activeBrokers)
          {
             boolean generic = false;

             List<BrokerParameters> brokers = new LinkedList<>();

             if (user != null)
             {
                // get a list of brokers for given account
                brokers = getBrokersForUser(user);
                if (brokers.isEmpty())
                {
                   generic = true;
                   // if no brokers are found for the given user get the list of
                   // brokers having no accounts
                   brokers = getBrokersForUser(null);
                }
             }

             return new Brokers(generic, brokers);
          }
       }

       /**
        * Removes user hosts associated with the broker agent that closes its session.
        *
        * @param brokerAgent
        */
       static void deregisterBroker(BrokerParameters brokerAgent)
       {
          String user = brokerAgent.getAssociatedUser();

          String hostNameOrIpAddress = brokerAgent.getAgentAddress();

          removeUserHostWithCheck(user, hostNameOrIpAddress, allHosts);

          removeUserHostWithCheck(user, hostNameOrIpAddress, dedicatedHosts);
       }

       /**
        * Removes a given user host from the target "user to hosts" map.
        *
        * @param    user
        *           The user account name
        * @param    hostNameOrIpAddress
        *           The agent host address
        * @param    targetUserHosts
        *           The target "user to hosts" map
        */
       private static void removeUserHostWithCheck(String user,
                                                   String hostNameOrIpAddress,
                                                   Map<String, List<String>> targetUserHosts)
       {
          List<String> userHosts = targetUserHosts.get(user);
          userHosts.remove(hostNameOrIpAddress);
          if (userHosts.isEmpty())
          {
             targetUserHosts.remove(user);
          }
       }

       /**
        * Select a broker from the broker list which has the lowest system loading.
        *
        * @param    brokers
        *           List of brokers.
        * @param    dedicatedUser
        *           The dedicated user for which the spawned client will be run
        * @param    clientAddress
        *           The client address
        *
        * @return  Scheduled broker.
        */
       private static BrokerParameters schedule(final List<BrokerParameters> brokers,
                                                String dedicatedUser,
                                                String clientAddress)
       {
          // system loading
          double minSystemLoading = Double.MAX_VALUE;

          // scheduled broker
          BrokerParameters scheduledBroker = null;

          try
          {
             InetAddress.getByName(clientAddress);
          }
          catch (UnknownHostException e)
          {
             throw new IllegalArgumentException("IP address of a host could not be determined " + clientAddress);
          }

          List<String> dedicatedHosts;

          dedicatedHosts = getHostsForUser(dedicatedUser, true);

          if (dedicatedHosts.isEmpty())
          {
             dedicatedHosts = getHostsForUser(dedicatedUser, false);
          }

          if (dedicatedHosts.contains(clientAddress))
          {
             dedicatedHosts = Collections.singletonList(clientAddress);
          }

          boolean anyHost = dedicatedHosts.isEmpty();

          // search for an available broker
          for (BrokerParameters broker : brokers)
          {
             try
             {
                // is broker still open?
                if (broker.getState() == BrokerState.OPEN)
                {
                   if (anyHost || dedicatedHosts.contains(broker.getRemote().getDedicatedHost()))
                   {
                      double systemLoading = broker.getRemote().getSystemLoading();
                      if (systemLoading < minSystemLoading)
                      {
                         minSystemLoading = systemLoading;
                         scheduledBroker = broker;
                      }
                   }
                }
             }
             catch (Throwable t)
             {
                LOG.logp(Level.SEVERE,
                         "BrokerManager.schedule()",
                         "",
                         "Remote broker call exception.",
                         t);
             }
          }

          // Check if broker has been scheduled.
          if (scheduledBroker == null)
          {
             throw new NullPointerException("No brokers are available for schedule.");
          }

          return scheduledBroker;
       }

       /**
        * Add a new user host to the target "user to hosts" map. If the same host address is already
        * in this map, then new RuntimeException is thrown.
        *
        * @param    user
        *           The user account name
        * @param    hostNameOrIpAddress
        *           The agent host address
        * @param    targetUserHosts
        *           The target "user to hosts" map
        */
       private static void addUserHostWithCheck(String user,
                                                String hostNameOrIpAddress,
                                                Map<String, List<String>> targetUserHosts)
       {
          List<String> userHosts = targetUserHosts.get(user);

          if (userHosts == null)
          {
             userHosts = new LinkedList<String>();
             targetUserHosts.put(user, userHosts);
          }

          if (userHosts.contains(hostNameOrIpAddress))
          {
             throw new RuntimeException("The broker agent has been already registered.");
          }

          userHosts.add(hostNameOrIpAddress);
       }

       /**
        * Get a list of brokers for a specified account.
        *
        * @param   user
        *          User account. If null select a list of brokers having no accounts.
        *
        * @return  A list of brokers which serve the user account or an empty list
        *          if no brokers are found.
        */
       private static List<BrokerParameters> getBrokersForUser(String user)
       {
          List<BrokerParameters> brokers = new LinkedList<>();

          for (Map.Entry<String, Broker> entry : activeBrokers.entrySet())
          {
             Broker broker = entry.getValue();
             // get brokers
             if (user == null ? broker.noAccounts() : broker.hasAccount(user))
             {
                Collection<BrokerParameters> values = broker.getBrokers().values();
                // select only open brokers
                for (BrokerParameters value : values)
                {
                   if (BrokerState.OPEN == value.getState())
                   {
                      brokers.add(value);
                   }
                }
             }
          }

          return brokers;
       }

       /**
        * Get dedicated hosts or all hosts for a specified account.
        *
        * @param   user
        *          The specified user account
        * @param   dedicatedMode
        *          True value indicates that only hosts that are for the dedicated broker clients,
        *          otherwise all user
        *
        * @return  A list of dedicated hosts for the user account or an empty list
        *          if no hosts are assigned.
        */
       private static List<String> getHostsForUser(String user, boolean dedicatedMode)
       {
          if (user == null)
          {
             return Collections.emptyList();
          }

          Map<String, List<String>> targetUserHosts;

          if (dedicatedMode)
          {
             targetUserHosts = dedicatedHosts;
          }
          else
          {
             targetUserHosts = allHosts;
          }

          synchronized (activeBrokers)
          {
             List<String> hosts = targetUserHosts.get(user);

             if (hosts == null)
             {
                return Collections.emptyList();
             }

             return new ArrayList<String>(hosts);
          }
       }

       /**
        * Find a broker using the given ID.
        *
        * @param   uuid
        *          Broker ID.
        *
        * @return  Broker parameters.
        */
       private static BrokerParameters findBroker(String uuid)
       {
          synchronized (activeBrokers)
          {
             if (uuid != null)
             {
                for (Map.Entry<String, Broker> entry : activeBrokers.entrySet())
                {
                   Broker broker = entry.getValue();

                   if (broker.findBroker(uuid))
                   {
                      return broker.getBroker(uuid);
                   }
                }
             }

             return null;
          }
       }

       /**
        * Remove a broker from registry when the remote session is closed.
        * The brokers are automatically unregistered when they close the session.
        *
        * @param   session
        *          Broker session instance.
        */
       private static void removeBroker(Session session)
       {
          synchronized (activeBrokers)
          {
             for (Map.Entry<String, Broker> entry : activeBrokers.entrySet())
             {
                Broker broker = entry.getValue();
                if (broker.removeBroker(session))
                {
                   return;
                }
             }
          }
       }

       /**
        * Get a list of P2J accounts which are served by this broker or and empty list if
        * no accounts are configured for this broker.
        *
        * @param   ds
        *          Directory services API.
        * @param   broker
        *          Directory path of this broker.
        *
        * @return  A list of accounts defined for this broker or an empty list if no accounts
        *          are configured for this broker.
        */
       private static List<String> getBrokerAccounts(DirectoryService ds, String broker)
       {
          List<String> accounts = new LinkedList<String>();

          Attribute[] attributes = ds.getNodeAttributes(broker);
          if (attributes != null)
          {
             for (Attribute attribute : attributes)
             {
                if ("account".equals(attribute.getName()))
                {
                   String[] values = attribute.getValues();
                   if (values != null && values.length > 0)
                   {
                      accounts.addAll(Arrays.asList(values));
                   }
                }
             }
          }

          return accounts;
       }

       /**
        * Update the command line parameters for the spawning process with the web client configuration.
        *
        * @param    args
        *           The broker spawn parameters to update
        * @param    allocatedResources
        *           The allocated web resources
        */
       private static void updateWebClientOptions(BrokerSpawnParameters args,
                                                  WebAllocatedResources allocatedResources)
       {
          if (allocatedResources != null)
          {
             int port = allocatedResources.getPort();
             if (port > 0)
             {
                args.getCommand().add(ConfigItem.PORT.fullOption() + "=" + port);
             }

             String forwardedHost = allocatedResources.getForwardedHost();
             if (forwardedHost != null)
             {
                args.getCommand().add(ConfigItem.PROXY_PROTOCOL.fullOption() + "=" + allocatedResources.getForwardedProto());
                args.getCommand().add(ConfigItem.PROXY_HOST.fullOption() + "=" + forwardedHost);
                args.getCommand().add(ConfigItem.WEB_ROOT.fullOption() + "=" + allocatedResources.getWebRoot());
             }
          }
       }
    }