<commit_before>/*
 * Copyright (c) 2014,2018 Red Hat.
 *
 * This library is free software; you can redistribute it and/or modify it
 * under the terms of the GNU Lesser General Public License as published
 * by the Free Software Foundation; either version 2.1 of the License, or
 * (at your option) any later version.
 *
 * This library 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 Lesser General Public
 * License for more details.
 */
#include "pmapi.h"
#include "libpcp.h"
#define SOCKET_INTERNAL
#include "internal.h"
#include "shellprobe.h"
#include <ctype.h>

#define PROBE	"__pmShellProbeDiscoverServices"

/*
 * Service discovery by indirect probing. The list of hostnames and/or IP
 * addresses to probe is determined by $PCP_BINADM_DIR/discover scripts.
 */
typedef struct targetInfo {
    char		*target;
    __pmHostEnt		*servInfo;	
} targetInfo;

typedef struct connectionOptions {
    char		path[MAXPATHLEN]; /* Discovery script path */
    char		cmd[128];	/* optional individual command */
    targetInfo		*targets;
    unsigned		numTargets;	/* Count of target addresses */
    unsigned		maxTargets;	/* High-water mark for targets */
    unsigned		maxThreads;	/* Max number of threads to use */
    struct timeval	timeout;	/* Connection timeout */
    const __pmServiceDiscoveryOptions *globalOptions; /* Global options */
} connectionOptions;

/* Context for each thread. */
typedef struct connectionContext {
    const char		*service;	/* Service spec */
    targetInfo		*nextTarget;	/* Next available target address */
    int			nports;		/* Number of ports per address */
    int			portIx;		/* Index of next available port */
    const int		*ports;		/* The actual ports */
    int			*numUrls;	/* Size of the results */
    char		***urls;	/* The results */
    const connectionOptions *options;	/* Connection options */
#if PM_MULTI_THREAD
    __pmMutex		addrLock;	/* lock for the above address/port */
    __pmMutex		urlLock;	/* lock for the above results */
#endif
} connectionContext;

/*
 * Attempt connection based on the given context until there are no more
 * addresses+ports to try.
 */
static void *
attemptConnections(void *arg)
{
    int			s;
    int			flags;
    int			sts;
    __pmFdSet		wfds;
    __pmServiceInfo	serviceInfo;
    __pmSockAddr	*addr;
    targetInfo		target;
    void		*enumIx;
    const __pmAddrInfo	*ai;
    const __pmServiceDiscoveryOptions *globalOptions;
    int			port;
    int			attempt;
    struct timeval	againWait = {0, 100000}; /* 0.1 seconds */
    connectionContext	*context = arg;

    /*
     * Keep trying to secure an address+port until there are no more
     * or until we are interrupted.
     */
    globalOptions = context->options->globalOptions;
    while (! globalOptions->timedOut &&
	   (! globalOptions->flags ||
	    (*globalOptions->flags & PM_SERVICE_DISCOVERY_INTERRUPTED) == 0)) {
	/* Obtain the address lock while securing the next target, if any. */
	PM_LOCK(context->addrLock);
	if (context->nextTarget == NULL) {
	    /* No more target addresses to probe. */
	    PM_UNLOCK(context->addrLock);
	    break;
	}

	/*
	 * There is an address+port remaining. Copy them locally.
	 */
	memcpy(&target, context->nextTarget, sizeof(target));
	port = context->ports[context->portIx];

	/*
	 * Advance the port index for the next thread. If we took the
	 * final port, then advance the address and reset the port index.
	 * The next target will then be set NULL which is the signal for
	 * all threads to exit.
	 */
	++context->portIx;
	if (context->portIx == context->nports) {
	    context->portIx = 0;
	    s = context->nextTarget - context->options->targets + 1;
	    if (context->options->numTargets <= s)
		context->nextTarget = NULL;
	    else
		context->nextTarget++;
	}
	PM_UNLOCK(context->addrLock);

	enumIx = NULL;
	for (addr = __pmHostEntGetSockAddr(target.servInfo, &enumIx);
	     addr != NULL;
	     addr = __pmHostEntGetSockAddr(target.servInfo, &enumIx)) {
	    /*
	     * We should be ignoring any entry without SOCK_STREAM (= 1),
	     * since we're connecting with TCP only. The other entries
	     * (SOCK_RAW, SOCK_DGRAM) are not relevant.
	     */
	    ai = __pmHostEntGetAddrInfo(target.servInfo, enumIx);
	    if (ai->ai_socktype != SOCK_STREAM) {
		__pmSockAddrFree(addr);
		continue;
	    }
	    __pmSockAddrSetPort(addr, port);

	    /*
	     * Create a socket. There is a limit on open fds, not just from
	     * the OS, but also in the IPC table. If we get EAGAIN, then
	     * wait 0.1 seconds and try again. We must have a limit in case
	     * something goes wrong. Make it 5 seconds (50 * 100,000 usecs).
	     */
	    for (attempt = 0; attempt < 50; ++attempt) {
		if (__pmSockAddrIsInet(addr))
		    s = __pmCreateSocket();
		else if (__pmSockAddrIsIPv6(addr))
		    s = __pmCreateIPv6Socket();
		else
		    s = -EINVAL;
		if (s != -EAGAIN)
		    break;
		__pmtimevalSleep(againWait);
	    }

	    if (pmDebugOptions.discovery && attempt > 0) {
		pmNotifyErr(LOG_INFO, "%s: Waited %d times for available fd\n",
				PROBE, attempt);
	    }

	    if (s < 0) {
	        pmNotifyErr(LOG_WARNING, "%s: Cannot create socket for %s:%d",
				PROBE, target.target, port);
		__pmSockAddrFree(addr);
		continue;
	    }

	    /*
	     * Attempt to connect. If flags comes back as less than zero,
	     * then the socket has already been closed by __pmConnectTo().
	     */
	    sts = -1;
	    flags = __pmConnectTo(s, addr, port);
	    if (flags >= 0) {
		/*
		 * FNDELAY and we're in progress - wait on __pmSelectWrite.
		 * __pmSelectWrite may alter the contents of the timeout so
		 * make a copy.
		 */
		struct timeval timeout = context->options->timeout;
		__pmFD_ZERO(&wfds);
		__pmFD_SET(s, &wfds);
		sts = __pmSelectWrite(s+1, &wfds, &timeout);

		/* Was the connection successful? */
		if (sts == 0)
		    sts = -1; /* Timed out */
		else if (sts > 0)
		    sts = __pmConnectCheckError(s);
		__pmCloseSocket(s);
	    }

	    /* If connection was successful, add this service to the list.  */
	    if (sts == 0) {
		serviceInfo.spec = context->service;
		serviceInfo.address = addr;
		if (strcmp(context->service, PM_SERVER_SERVICE_SPEC) == 0)
		    serviceInfo.protocol = SERVER_PROTOCOL;
		else if (strcmp(context->service, PM_SERVER_PROXY_SPEC) == 0)
		    serviceInfo.protocol = PROXY_PROTOCOL;
		else if (strcmp(context->service, PM_SERVER_WEBD_SPEC) == 0)
		    serviceInfo.protocol = PMWEBD_PROTOCOL;

		PM_LOCK(context->urlLock);
		*context->numUrls =
			__pmAddDiscoveredService(&serviceInfo, globalOptions,
					 *context->numUrls, context->urls);
		PM_UNLOCK(context->urlLock);
	    }

	    __pmSockAddrFree(addr);
	}
    }

    return NULL;
}

static int
shellProbeForServices(const char *script, const char *service,
    const connectionOptions *options, int numUrls, char ***urls)
{
    int			*ports = NULL;
    int			nports;
    int			prevNumUrls = numUrls;
    connectionContext	context;
#if PM_MULTI_THREAD
    int			sts;
    pthread_t		*threads = NULL;
    unsigned		threadIx;
    unsigned		nThreads;
    pthread_attr_t	threadAttr;
#endif

    /* Determine the port numbers for this service. */
    ports = NULL;
    nports = 0;
    nports = __pmServiceAddPorts(service, &ports, nports);
    if (nports <= 0) {
	pmNotifyErr(LOG_ERR, "%s: could not find ports for service '%s'",
		      PROBE, service);
	return 0;
    }

    /*
     * Initialize the shared probing context. This will be shared among all of
     * the worker threads.
     */
    context.service = service;
    context.ports = ports;
    context.nports = nports;
    context.numUrls = &numUrls;
    context.urls = urls;
    context.portIx = 0;
    context.options = options;
    context.nextTarget = &options->targets[0];

#if PM_MULTI_THREAD
    /*
     * Set up the concurrrency controls. These locks will be tested
     * even if we fail to allocate/use the thread table below.
     */
    pthread_mutex_init(&context.addrLock, NULL);
    pthread_mutex_init(&context.urlLock, NULL);

    if (options->maxThreads > 0) {<commit_msg>qa: fix 724, debug-conditional unknown probe port diagnostic<commit_after>/*
 * Copyright (c) 2014,2018 Red Hat.
 *
 * This library is free software; you can redistribute it and/or modify it
 * under the terms of the GNU Lesser General Public License as published
 * by the Free Software Foundation; either version 2.1 of the License, or
 * (at your option) any later version.
 *
 * This library 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 Lesser General Public
 * License for more details.
 */
#include "pmapi.h"
#include "libpcp.h"
#define SOCKET_INTERNAL
#include "internal.h"
#include "shellprobe.h"
#include <ctype.h>

#define PROBE	"__pmShellProbeDiscoverServices"

/*
 * Service discovery by indirect probing. The list of hostnames and/or IP
 * addresses to probe is determined by $PCP_BINADM_DIR/discover scripts.
 */
typedef struct targetInfo {
    char		*target;
    __pmHostEnt		*servInfo;	
} targetInfo;

typedef struct connectionOptions {
    char		path[MAXPATHLEN]; /* Discovery script path */
    char		cmd[128];	/* optional individual command */
    targetInfo		*targets;
    unsigned		numTargets;	/* Count of target addresses */
    unsigned		maxTargets;	/* High-water mark for targets */
    unsigned		maxThreads;	/* Max number of threads to use */
    struct timeval	timeout;	/* Connection timeout */
    const __pmServiceDiscoveryOptions *globalOptions; /* Global options */
} connectionOptions;

/* Context for each thread. */
typedef struct connectionContext {
    const char		*service;	/* Service spec */
    targetInfo		*nextTarget;	/* Next available target address */
    int			nports;		/* Number of ports per address */
    int			portIx;		/* Index of next available port */
    const int		*ports;		/* The actual ports */
    int			*numUrls;	/* Size of the results */
    char		***urls;	/* The results */
    const connectionOptions *options;	/* Connection options */
#if PM_MULTI_THREAD
    __pmMutex		addrLock;	/* lock for the above address/port */
    __pmMutex		urlLock;	/* lock for the above results */
#endif
} connectionContext;

/*
 * Attempt connection based on the given context until there are no more
 * addresses+ports to try.
 */
static void *
attemptConnections(void *arg)
{
    int			s;
    int			flags;
    int			sts;
    __pmFdSet		wfds;
    __pmServiceInfo	serviceInfo;
    __pmSockAddr	*addr;
    targetInfo		target;
    void		*enumIx;
    const __pmAddrInfo	*ai;
    const __pmServiceDiscoveryOptions *globalOptions;
    int			port;
    int			attempt;
    struct timeval	againWait = {0, 100000}; /* 0.1 seconds */
    connectionContext	*context = arg;

    /*
     * Keep trying to secure an address+port until there are no more
     * or until we are interrupted.
     */
    globalOptions = context->options->globalOptions;
    while (! globalOptions->timedOut &&
	   (! globalOptions->flags ||
	    (*globalOptions->flags & PM_SERVICE_DISCOVERY_INTERRUPTED) == 0)) {
	/* Obtain the address lock while securing the next target, if any. */
	PM_LOCK(context->addrLock);
	if (context->nextTarget == NULL) {
	    /* No more target addresses to probe. */
	    PM_UNLOCK(context->addrLock);
	    break;
	}

	/*
	 * There is an address+port remaining. Copy them locally.
	 */
	memcpy(&target, context->nextTarget, sizeof(target));
	port = context->ports[context->portIx];

	/*
	 * Advance the port index for the next thread. If we took the
	 * final port, then advance the address and reset the port index.
	 * The next target will then be set NULL which is the signal for
	 * all threads to exit.
	 */
	++context->portIx;
	if (context->portIx == context->nports) {
	    context->portIx = 0;
	    s = context->nextTarget - context->options->targets + 1;
	    if (context->options->numTargets <= s)
		context->nextTarget = NULL;
	    else
		context->nextTarget++;
	}
	PM_UNLOCK(context->addrLock);

	enumIx = NULL;
	for (addr = __pmHostEntGetSockAddr(target.servInfo, &enumIx);
	     addr != NULL;
	     addr = __pmHostEntGetSockAddr(target.servInfo, &enumIx)) {
	    /*
	     * We should be ignoring any entry without SOCK_STREAM (= 1),
	     * since we're connecting with TCP only. The other entries
	     * (SOCK_RAW, SOCK_DGRAM) are not relevant.
	     */
	    ai = __pmHostEntGetAddrInfo(target.servInfo, enumIx);
	    if (ai->ai_socktype != SOCK_STREAM) {
		__pmSockAddrFree(addr);
		continue;
	    }
	    __pmSockAddrSetPort(addr, port);

	    /*
	     * Create a socket. There is a limit on open fds, not just from
	     * the OS, but also in the IPC table. If we get EAGAIN, then
	     * wait 0.1 seconds and try again. We must have a limit in case
	     * something goes wrong. Make it 5 seconds (50 * 100,000 usecs).
	     */
	    for (attempt = 0; attempt < 50; ++attempt) {
		if (__pmSockAddrIsInet(addr))
		    s = __pmCreateSocket();
		else if (__pmSockAddrIsIPv6(addr))
		    s = __pmCreateIPv6Socket();
		else
		    s = -EINVAL;
		if (s != -EAGAIN)
		    break;
		__pmtimevalSleep(againWait);
	    }

	    if (pmDebugOptions.discovery && attempt > 0) {
		pmNotifyErr(LOG_INFO, "%s: Waited %d times for available fd\n",
				PROBE, attempt);
	    }

	    if (s < 0) {
	        pmNotifyErr(LOG_WARNING, "%s: Cannot create socket for %s:%d",
				PROBE, target.target, port);
		__pmSockAddrFree(addr);
		continue;
	    }

	    /*
	     * Attempt to connect. If flags comes back as less than zero,
	     * then the socket has already been closed by __pmConnectTo().
	     */
	    sts = -1;
	    flags = __pmConnectTo(s, addr, port);
	    if (flags >= 0) {
		/*
		 * FNDELAY and we're in progress - wait on __pmSelectWrite.
		 * __pmSelectWrite may alter the contents of the timeout so
		 * make a copy.
		 */
		struct timeval timeout = context->options->timeout;
		__pmFD_ZERO(&wfds);
		__pmFD_SET(s, &wfds);
		sts = __pmSelectWrite(s+1, &wfds, &timeout);

		/* Was the connection successful? */
		if (sts == 0)
		    sts = -1; /* Timed out */
		else if (sts > 0)
		    sts = __pmConnectCheckError(s);
		__pmCloseSocket(s);
	    }

	    /* If connection was successful, add this service to the list.  */
	    if (sts == 0) {
		serviceInfo.spec = context->service;
		serviceInfo.address = addr;
		if (strcmp(context->service, PM_SERVER_SERVICE_SPEC) == 0)
		    serviceInfo.protocol = SERVER_PROTOCOL;
		else if (strcmp(context->service, PM_SERVER_PROXY_SPEC) == 0)
		    serviceInfo.protocol = PROXY_PROTOCOL;
		else if (strcmp(context->service, PM_SERVER_WEBD_SPEC) == 0)
		    serviceInfo.protocol = PMWEBD_PROTOCOL;

		PM_LOCK(context->urlLock);
		*context->numUrls =
			__pmAddDiscoveredService(&serviceInfo, globalOptions,
					 *context->numUrls, context->urls);
		PM_UNLOCK(context->urlLock);
	    }

	    __pmSockAddrFree(addr);
	}
    }

    return NULL;
}

static int
shellProbeForServices(const char *script, const char *service,
    const connectionOptions *options, int numUrls, char ***urls)
{
    int			*ports = NULL;
    int			nports;
    int			prevNumUrls = numUrls;
    connectionContext	context;
#if PM_MULTI_THREAD
    int			sts;
    pthread_t		*threads = NULL;
    unsigned		threadIx;
    unsigned		nThreads;
    pthread_attr_t	threadAttr;
#endif

    /* Determine the port numbers for this service. */
    ports = NULL;
    nports = 0;
    nports = __pmServiceAddPorts(service, &ports, nports);
    if (nports <= 0) {
	if (pmDebugOptions.discovery)
	    pmNotifyErr(LOG_INFO, "%s: no recognised ports for service '%s'",
			PROBE, service);
	return 0;
    }

    /*
     * Initialize the shared probing context. This will be shared among all of
     * the worker threads.
     */
    context.service = service;
    context.ports = ports;
    context.nports = nports;
    context.numUrls = &numUrls;
    context.urls = urls;
    context.portIx = 0;
    context.options = options;
    context.nextTarget = &options->targets[0];

#if PM_MULTI_THREAD
    /*
     * Set up the concurrrency controls. These locks will be tested
     * even if we fail to allocate/use the thread table below.
     */
    pthread_mutex_init(&context.addrLock, NULL);
    pthread_mutex_init(&context.urlLock, NULL);

    if (options->maxThreads > 0) {
{'fraction_non_alphanumeric': 0.10511554173526004, 'fraction_numerical': 0.0074995427108103165, 'mean_word_length': 2.9936693450206966, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 87, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '896404', 'n_tokens_mistral': 5613, 'n_tokens_neox': 5151, 'n_words': 1648}