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());
}
}
}
}