http

Documentation

Index

• Variables
• func NewUpstream(poolManager *pool.Manager) upstream.Upstream
• type Upstream
  • func (u *Upstream) CheckHealth(ctx context.Context) error
  • func (u *Upstream) Register(ctx context.Context, serviceConfig *configv1.UpstreamServiceConfig, ...) (string, []*configv1.ToolDefinition, []*configv1.ResourceDefinition, error)
  • func (u *Upstream) Shutdown(_ context.Context) error

Variables

var NewHTTPPool = func(
	minSize, maxSize int,
	idleTimeout time.Duration,
	config *configv1.UpstreamServiceConfig,
) (pool.Pool[*client.HTTPClientWrapper], error) {
	tlsConfig := &tls.Config{
		MinVersion:         tls.VersionTLS12,
		InsecureSkipVerify: config.GetHttpService().GetTlsConfig().GetInsecureSkipVerify(),
	}

	if mtlsConfig := config.GetUpstreamAuth().GetMtls(); mtlsConfig != nil {
		if err := validation.IsSecurePath(mtlsConfig.GetClientCertPath()); err != nil {
			return nil, fmt.Errorf("invalid client certificate path: %w", err)
		}
		if err := validation.IsSecurePath(mtlsConfig.GetClientKeyPath()); err != nil {
			return nil, fmt.Errorf("invalid client key path: %w", err)
		}
		cert, err := tls.LoadX509KeyPair(mtlsConfig.GetClientCertPath(), mtlsConfig.GetClientKeyPath())
		if err != nil {
			return nil, err
		}
		tlsConfig.Certificates = []tls.Certificate{cert}

		if err := validation.IsSecurePath(mtlsConfig.GetCaCertPath()); err != nil {
			return nil, fmt.Errorf("invalid CA certificate path: %w", err)
		}
		caCert, err := os.ReadFile(mtlsConfig.GetCaCertPath())
		if err != nil {
			return nil, err
		}
		caCertPool := x509.NewCertPool()
		caCertPool.AppendCertsFromPEM(caCert)
		tlsConfig.RootCAs = caCertPool
	}

	dialer := util.NewSafeDialer()

	if os.Getenv("MCPANY_DANGEROUS_ALLOW_LOCAL_IPS") == util.TrueStr {
		dialer.AllowLoopback = true
		dialer.AllowPrivate = true
	}

	if os.Getenv("MCPANY_ALLOW_LOOPBACK_RESOURCES") == util.TrueStr {
		dialer.AllowLoopback = true
	}

	if os.Getenv("MCPANY_ALLOW_PRIVATE_NETWORK_RESOURCES") == util.TrueStr {
		dialer.AllowPrivate = true
	}

	baseTransport := &http.Transport{
		TLSClientConfig:     tlsConfig,
		DialContext:         dialer.DialContext,
		MaxIdleConns:        maxSize,
		MaxIdleConnsPerHost: maxSize,

		IdleConnTimeout:       90 * time.Second,
		TLSHandshakeTimeout:   10 * time.Second,
		ExpectContinueTimeout: 1 * time.Second,
		ForceAttemptHTTP2:     true,
	}

	clientTimeout := 30 * time.Second
	if config.GetResilience() != nil && config.GetResilience().GetTimeout() != nil {
		clientTimeout = config.GetResilience().GetTimeout().AsDuration()
	}

	sharedClient := &http.Client{
		Transport: otelhttp.NewTransport(baseTransport),
		Timeout:   clientTimeout,
	}

	checker := healthChecker.NewChecker(config)

	factory := func(_ context.Context) (*client.HTTPClientWrapper, error) {
		return client.NewHTTPClientWrapper(
			sharedClient,
			config,
			checker,
		), nil
	}

	basePool, err := pool.New(factory, minSize, minSize, maxSize, idleTimeout, false)
	if err != nil {
		return nil, err
	}

	return &httpPool{
		Pool:      basePool,
		transport: baseTransport,
	}, nil
}

NewHTTPPool creates a new connection pool for HTTP clients.

It is defined as a variable to allow for easy mocking in tests.

Parameters:

• minSize (int): The initial number of clients to create.
• maxSize (int): The maximum number of clients the pool can hold.
• idleTimeout (time.Duration): The duration after which an idle client may be closed.
• config (*configv1.UpstreamServiceConfig): The configuration for the upstream service.

Returns:

• pool.Pool[*client.HTTPClientWrapper]: The created pool.
• error: An error if the pool cannot be created.

Errors:

• Returns error if TLS configuration is invalid (e.g., certificate files missing).
• Returns error if pool creation fails.

Side Effects:

• Reads certificate files if mTLS is configured.
• Initializes a new http.Transport and http.Client.

Summary: Represents a NewHTTPPool.

Functions

func NewUpstream

func NewUpstream(poolManager *pool.Manager) upstream.Upstream

NewUpstream creates a new instance of Upstream.

Parameters:

• poolManager (*pool.Manager): The connection pool manager to be used for managing HTTP connections.

Returns:

• upstream.Upstream: An implementation of the upstream.Upstream interface.

Side Effects:

• Allocates memory for the Upstream struct.

Summary: Initializes NewUpstream operation.

Parameters:

• TODO: Document parameters.

Returns:

• TODO: Document returns.

Errors:

• TODO: Document errors.

Side Effects:

• None.

Types

type Upstream

type Upstream struct {
	// contains filtered or unexported fields
}

Upstream implements the upstream.Upstream interface for services that are exposed via standard HTTP endpoints.

It handles the registration of tools defined in the service configuration and manages connection pooling for HTTP requests.

Summary: Represents a Upstream.

func (*Upstream) CheckHealth

func (u *Upstream) CheckHealth(ctx context.Context) error

CheckHealth performs a health check on the upstream service.

It uses a configured health checker if available, or falls back to a basic TCP connection check to the service address.

Parameters:

• ctx (context.Context): The context for the health check.

Returns:

• error: An error if the service is unhealthy or unreachable.

Errors:

• Returns an error if the health check fails or the address is not configured.

Side Effects:

• May establish a network connection to the service.

Summary: Executes CheckHealth operation.

Parameters:

• TODO: Document parameters.

Returns:

• TODO: Document returns.

Errors:

• TODO: Document errors.

Side Effects:

• None.

func (*Upstream) Register

func (u *Upstream) Register(
	ctx context.Context,
	serviceConfig *configv1.UpstreamServiceConfig,
	toolManager tool.ManagerInterface,
	promptManager prompt.ManagerInterface,
	resourceManager resource.ManagerInterface,
	isReload bool,
) (string, []*configv1.ToolDefinition, []*configv1.ResourceDefinition, error)

Register processes the configuration for an HTTP service, creates a connection pool for it, and then creates and registers tools for each call definition specified in the configuration.

Parameters:

• ctx (context.Context): The context for the registration process.
• serviceConfig (*configv1.UpstreamServiceConfig): The configuration for the upstream service.
• toolManager (tool.ManagerInterface): The manager where discovered tools will be registered.
• promptManager (prompt.ManagerInterface): The manager where discovered prompts will be registered.
• resourceManager (resource.ManagerInterface): The manager where discovered resources will be registered.
• isReload (bool): Indicates whether this is a configuration reload.

Returns:

• string: The unique service ID.
• []*configv1.ToolDefinition: A list of registered tool definitions.
• []*configv1.ResourceDefinition: A list of registered resource definitions.
• error: An error if registration fails.

Errors:

• Returns error if service configuration is invalid (e.g., missing address).
• Returns error if connection pool creation fails.

Side Effects:

• Creates and registers a new HTTP connection pool.
• Starts a health checker for the service.
• Registers tools and prompts with their respective managers.

Summary: Executes Register operation.

Parameters:

• TODO: Document parameters.

Returns:

• TODO: Document returns.

Errors:

• TODO: Document errors.

Side Effects:

• None.

func (*Upstream) Shutdown

func (u *Upstream) Shutdown(_ context.Context) error

Shutdown gracefully terminates the HTTP upstream service by shutting down the associated connection pool.

Parameters:

• ctx (context.Context): The context for the shutdown operation (currently unused).

Returns:

• error: Always returns nil.

Side Effects:

• Stops the health checker.
• Deregisters the connection pool.

Summary: Executes Shutdown operation.

Parameters:

• TODO: Document parameters.

Returns:

• TODO: Document returns.

Errors:

• TODO: Document errors.

Side Effects:

• None.