Set the opened ports of a charm

doc
1

See first: Juju open-port

Use at least Juju 2.9 for machine charms, and at least Juju 3.1 for Kubernetes charms.

Contents:

  1. Open a port based on the charm config
  2. Test the port is opened

Open a port based on the charm config

To make a service permanently reachable under a stable URL on the cluster, the charm needs to open (expose) a port. This ensures that the charm will be consistently accessible even if the pod gets recycled and the IP address changes.

The port that the charm’s service should be reachable on is typically defined as a config option. As such, all the usual procedure for adding a config option to a charm applies:

  1. In your charmcraft.yaml define a config that sets the port to be opened/exposed. For example:
options:
  config:
    server-port:
      type: int
      description: the port on which to offer the service
      default: 8000
  1. In your src/charm.py observe the config-changed event and define a handler. For example:
self.framework.observe(self.on.config_changed, self._on_config_changed)

Now, in the body of the charm definition, define the event handler, and set the port that should be open – this defaults to a TCP port, but UDP and ICMP can also be specified. For example:

def _on_config_changed(self, event: ops.ConfigChangedEvent):
    self.unit.set_ports(self.config["server-port"])

Examples: loki-k8s sets the open port in the charm __init__, mysql-router sets ports based on whether the charm is configured to be externally accessible

See more: How to add a config option to a charm, ops.Unit.set_ports

Test the port is opened

See first: Get started with charm testing

You’ll want to add three levels of tests: unit, scenario, and integration. For a charm that exposes the ports to open via config options, the tests are much the same as for testing adding config options, just that you must also make sure to verify that the port is opened.

Write unit tests

See first: How to write unit tests for a charm

To use a unit test to verify that updating the charm config opens a port, for this charm that opens a port based on the config, the test needs to trigger the config-changed event and then check which ports are open. In your tests/unit/test_charm.py file, add the following test function to the file:

def test_port_configuration():
    harness = ops.testing.Harness()
    harness.begin()

    port = 8080
    harness.update_config({"server-port": port})

    assert harness.model.unit.opened_ports() == {ops.Port("tcp", port)}

Examples: charm-microk8s checks that the install hook opens a port

See more: How to test a config-changed observer, ops.Unit.opened_ports

Write scenario tests

See first: How to write scenario tests for a charm

To use a scenario test to verify that the config-changed event results in the port being opened, pass the new config to the State, and, after running the event, check the State.opened_ports attribute. For example, in your tests/scenario/test_charm.py file, add the following test function:

def test_open_port():
    ctx = scenario.Context(MyCharm)

    port = 8080
    state_out = ctx.run("config_changed", scenario.State(config={"server-port": port}))

    assert len(state.opened_ports) == 1
    assert state_out.opened_ports[0].port == port
    assert state_out.opened_ports[0].protocol == "tcp"

See more: scenario.Ports

Write integration tests

See first: How to write integration tests for a charm

To verify that the port is opened when the charm is deployed to a Juju model, add an integration test that opens a real connection to the appropriate port. In your tests/integration/test_charm.py file, add helper methods that can be used to check if a port is open:

async def get_address(ops_test: OpsTest, app_name, unit_num=0) -> str:
    """Get the address for a the service for an app."""
    status = await ops_test.model.get_status()
    return status["applications"][app_name].get_public_address()

def is_port_open(host: str, port: int, timeout: float=5.0) -> bool:
    """Check if a port is opened on a particular host."""
    try:
        with socket.create_connection((host, port), timeout=timeout):
            # If the connection succeeds, the port is open.
    except (ConnectionRefusedError, TimeoutError):
        # If the connection fails, the port is not open.
        return False

See more: Unit.get_public_address

See more: socket.create_connection

Now add the test case that will trigger the config-changed event that will open the port:

@pytest.mark.abort_on_fail
async def test_open_ports(ops_test: OpsTest):
    app = ops_test.model.applications.get(APP_NAME)

    # Get the service address of the application:
    address = await get_address(ops_test=ops_test, app_name=APP_NAME)
    # Verify that the default port is opened:
    assert is_port_open(address, 8000)

    # Change the config to choose a different port, and verify that it is opened:
    new_port = 8001
    await app.set_config({"server-port": new_port})
    await ops_test.model.wait_for_idle(
        apps=[APP_NAME], status="active", timeout=600
    ),
    assert is_port_open(address, new_port)

Examples: loki-k8s verifies that the rules are externally accessible, mysql-router-k8s verifies that queries can be performed using the external address


Contributors:@adithya-raj, @mmkay, @ibraaoad, @tmihoc, @tony-meyer

1 Like