Kubernetes-REST
view release on metacpan or search on metacpan
lib/Kubernetes/REST.pm view on Meta::CPAN
my @events;
while ($$buffer_ref =~ s/^([^\n]*)\n//) {
my $line = $1;
next unless length $line;
my $data = eval { $self->_json->decode($line) };
next unless $data;
my $type = $data->{type} // '';
my $raw_object = $data->{object} // {};
# Track resourceVersion for resumability
my $rv;
if ($raw_object->{metadata} && $raw_object->{metadata}{resourceVersion}) {
$rv = $raw_object->{metadata}{resourceVersion};
}
# Inflate the object (ERROR events stay as hashrefs)
my $object;
if ($type eq 'ERROR') {
$object = $raw_object;
} else {
$object = eval { $self->k8s->struct_to_object($class, $raw_object) }
// $raw_object;
}
push @events, {
event => Kubernetes::REST::WatchEvent->new(
type => $type,
object => $object,
raw => $raw_object,
),
resourceVersion => $rv,
is_error => ($type eq 'ERROR' ? 1 : 0),
error_code => ($type eq 'ERROR' ? ($raw_object->{code} // 0) : 0),
};
}
return @events;
}
sub _process_log_chunk {
my ($self, $buffer_ref, $chunk) = @_;
$$buffer_ref .= $chunk;
my @events;
while ($$buffer_ref =~ s/^([^\n]*)\n//) {
my $line = $1;
push @events, Kubernetes::REST::LogEvent->new(line => $line);
}
return @events;
}
# ============================================================================
# PUBLIC BUILDING BLOCKS FOR ASYNC WRAPPERS
#
# These methods expose the internal request/response pipeline as a stable API
# for async wrappers (e.g. Net::Async::Kubernetes) that need to build requests,
# process responses, and handle streaming without going through the sync
# convenience methods (list, get, watch, log, port_forward, exec, attach, etc.).
# ============================================================================
sub build_path {
my ($self, @args) = @_;
return $self->_build_path(@args);
}
sub prepare_request {
my ($self, @args) = @_;
return $self->_prepare_request(@args);
}
sub check_response {
my ($self, @args) = @_;
return $self->_check_response(@args);
}
sub inflate_object {
my ($self, @args) = @_;
return $self->_inflate_object(@args);
}
sub inflate_list {
my ($self, @args) = @_;
return $self->_inflate_list(@args);
}
sub process_watch_chunk {
my ($self, @args) = @_;
return $self->_process_watch_chunk(@args);
}
sub process_log_chunk {
my ($self, @args) = @_;
return $self->_process_log_chunk(@args);
}
# Convenience: prepare + call in one step (used by sync CRUD methods)
sub _request {
my ($self, $method, $path, $body, %opts) = @_;
my $req = $self->_prepare_request($method, $path,
body => $body,
%opts,
);
return $self->io->call($req);
lib/Kubernetes/REST.pm view on Meta::CPAN
(my $kind = ref $obj) =~ s/.*:://;
my $key = join("\0", $kind, $obj->metadata->namespace // '');
$expected{$key}{$obj->metadata->name} = 1;
}
for my $kind (@kinds) {
for my $ns (@namespaces) {
my %list_args = (labelSelector => $label);
$list_args{namespace} = $ns if defined $ns;
my $list = eval { $self->list($kind, %list_args) };
next unless $list;
for my $item (@{$list->items}) {
my $item_ns = $item->metadata->namespace // '';
my $key = join("\0", $kind, $item_ns);
next if $expected{$key} && $expected{$key}{$item->metadata->name};
eval { $self->delete($item) };
}
}
}
return @results;
}
sub watch {
my ($self, $short_class, %args) = @_;
my $on_event = delete $args{on_event}
or croak "watch requires 'on_event' callback";
my $timeout = delete $args{timeout} // 300;
my $resource_version = delete $args{resourceVersion};
my $label_selector = delete $args{labelSelector};
my $field_selector = delete $args{fieldSelector};
my $class = $self->expand_class($short_class);
my $path = $self->_build_path($class, %args);
my %params = (
watch => 'true',
timeoutSeconds => $timeout,
);
$params{resourceVersion} = $resource_version if defined $resource_version;
$params{labelSelector} = $label_selector if defined $label_selector;
$params{fieldSelector} = $field_selector if defined $field_selector;
my $req = $self->_prepare_request('GET', $path, parameters => \%params);
my $buffer = '';
my $last_rv = $resource_version;
my $got_410 = 0;
my $data_callback = sub {
my ($chunk) = @_;
for my $result ($self->_process_watch_chunk($class, \$buffer, $chunk)) {
$last_rv = $result->{resourceVersion} if $result->{resourceVersion};
$got_410 = 1 if $result->{error_code} == 410;
$on_event->($result->{event});
}
};
my $response = $self->io->call_streaming($req, $data_callback);
$self->_check_response($response, "watch $short_class");
croak "Watch expired (410 Gone): resourceVersion too old, re-list to get a fresh resourceVersion"
if $got_410;
return $last_rv;
}
sub log {
my ($self, $short_class, @rest) = @_;
# Support: log('Pod', 'name', ...) and log('Pod', name => 'name', ...)
my %args;
if (@rest >= 1 && !ref($rest[0]) && $rest[0] !~ /^(name|namespace|container|follow|tailLines|sinceSeconds|sinceTime|timestamps|previous|limitBytes|on_line)$/) {
$args{name} = shift @rest;
%args = (%args, @rest);
} elsif (@rest % 2 == 0) {
%args = @rest;
} else {
croak "Invalid arguments to log()";
}
croak "name required for log" unless $args{name};
my $on_line = delete $args{on_line};
my $container = delete $args{container};
my $follow = delete $args{follow};
my $tail_lines = delete $args{tailLines};
my $since_seconds = delete $args{sinceSeconds};
my $since_time = delete $args{sinceTime};
my $timestamps = delete $args{timestamps};
my $previous = delete $args{previous};
my $limit_bytes = delete $args{limitBytes};
my $class = $self->expand_class($short_class);
my $path = $self->_build_path($class, %args) . '/log';
my %params;
$params{container} = $container if defined $container;
$params{follow} = 'true' if $follow;
$params{tailLines} = $tail_lines if defined $tail_lines;
$params{sinceSeconds} = $since_seconds if defined $since_seconds;
$params{sinceTime} = $since_time if defined $since_time;
$params{timestamps} = 'true' if $timestamps;
$params{previous} = 'true' if $previous;
$params{limitBytes} = $limit_bytes if defined $limit_bytes;
if ($on_line) {
# Streaming mode
my $req = $self->_prepare_request('GET', $path, parameters => \%params);
my $buffer = '';
my $data_callback = sub {
my ($chunk) = @_;
for my $event ($self->_process_log_chunk(\$buffer, $chunk)) {
$on_line->($event);
}
};
my $response = $self->io->call_streaming($req, $data_callback);
$self->_check_response($response, "log $short_class");
# Process any remaining data in buffer (last line without trailing newline)
if (length $buffer) {
$on_line->(Kubernetes::REST::LogEvent->new(line => $buffer));
}
return;
} else {
# One-shot mode
my $response = $self->_request('GET', $path, undef,
%params ? (parameters => \%params) : (),
);
$self->_check_response($response, "log $short_class");
return $response->content;
}
}
sub port_forward {
my ($self, $short_class, @rest) = @_;
my %args;
if (@rest >= 1 && !ref($rest[0]) && $rest[0] !~ /^(name|namespace|ports|subprotocol|on_open|on_frame|on_close|on_error)$/) {
$args{name} = shift @rest;
%args = (%args, @rest);
} elsif (@rest % 2 == 0) {
%args = @rest;
} else {
croak "Invalid arguments to port_forward()";
}
croak "name required for port_forward" unless $args{name};
my $ports = delete $args{ports};
croak "ports required for port_forward" unless defined $ports;
$ports = [$ports] unless ref($ports) eq 'ARRAY';
croak "ports required for port_forward" unless @$ports;
for my $p (@$ports) {
croak "invalid port '$p' for port_forward"
unless defined($p) && $p =~ /^\d+$/ && $p > 0 && $p <= 65535;
}
my $subprotocol = delete $args{subprotocol} // 'v4.channel.k8s.io';
my $on_open = delete $args{on_open};
my $on_frame = delete $args{on_frame};
my $on_close = delete $args{on_close};
my $on_error = delete $args{on_error};
my $class = $self->expand_class($short_class);
my $path = $self->_build_path($class, %args) . '/portforward';
my $req = $self->_prepare_request('GET', $path,
parameters => { ports => $ports },
headers => {
Accept => '*/*',
Connection => 'Upgrade',
Upgrade => 'websocket',
'Sec-WebSocket-Protocol' => $subprotocol,
lib/Kubernetes/REST.pm view on Meta::CPAN
Returns a hashref with the OpenAPI v2 schema definition.
=head2 compare_schema
my $result = $api->compare_schema('Pod');
Compare the local L<IO::K8s> class definition against the cluster's OpenAPI schema. Useful for detecting version skew between your L<IO::K8s> installation and the cluster.
Returns the comparison result from C<< IO::K8s::Resource->compare_to_schema >>.
=head2 build_path
my $class = $api->expand_class('Pod');
my $path = $api->build_path($class, name => 'my-pod', namespace => 'default');
# => /api/v1/namespaces/default/pods/my-pod
Build the REST API URL path for a resource class. Takes a fully-qualified class name (from L</expand_class>) and optional C<name>/C<namespace> arguments.
This is a public API for async wrappers like L<Net::Async::Kubernetes> that need to construct request paths independently.
=head2 prepare_request
my $req = $api->prepare_request('GET', $path,
parameters => \%params,
body => \%body,
);
Build a L<Kubernetes::REST::HTTPRequest> with method, full URL, authorization
headers, and optional query parameters or JSON body.
Query parameter values may be scalars or arrayrefs (arrayrefs are emitted as
repeated C<key=value> pairs). Extra request headers can be provided via
C<headers =E<gt> \%headers>.
This is a public API for async wrappers that execute HTTP requests through their own event loop.
=head2 check_response
$api->check_response($response, "get Pod");
Validate an HTTP response. Croaks with a descriptive error if the status code is >= 400. Returns the response on success.
=head2 inflate_object
my $pod = $api->inflate_object($class, $response);
Decode the JSON response body and inflate it into a typed L<IO::K8s> object.
=head2 inflate_list
my $list = $api->inflate_list($class, $response);
Decode the JSON response body and inflate the C<items> array into an L<IO::K8s::List> of typed objects.
=head2 process_watch_chunk
my @results = $api->process_watch_chunk($class, \$buffer, $chunk);
Process a chunk of NDJSON watch data. Appends the chunk to the buffer, extracts complete lines, and returns a list of hashrefs with C<event> (L<Kubernetes::REST::WatchEvent>), C<resourceVersion>, C<is_error>, and C<error_code>.
This is a public API for async wrappers that handle streaming watch responses through their own event loop.
=head2 process_log_chunk
my @events = $api->process_log_chunk(\$buffer, $chunk);
Process a chunk of plain-text log data. Appends the chunk to the buffer, extracts complete lines, and returns a list of L<Kubernetes::REST::LogEvent> objects.
This is a public API for async wrappers that handle streaming log responses through their own event loop.
=head2 list
my $list = $api->list('Pod', namespace => 'default');
my $list = $api->list('Namespace', labelSelector => 'app=web');
List resources. Returns an L<IO::K8s::List> object.
Accepts short class names (C<Pod>) or full class paths. For namespaced resources, pass C<namespace> parameter. Omit C<namespace> to list cluster-scoped resources.
Supports C<labelSelector> and C<fieldSelector> query parameters for server-side filtering.
=head2 get
my $pod = $api->get('Pod', name => 'my-pod', namespace => 'default');
# or shorthand:
my $pod = $api->get('Pod', 'my-pod', namespace => 'default');
Get a single resource by name. Returns a typed L<IO::K8s> object.
=head2 create
my $created = $api->create($pod);
Create a resource from an L<IO::K8s> object. Returns the created object with server-assigned fields (UID, resourceVersion, etc.).
=head2 update
my $updated = $api->update($pod);
Update an existing resource. Replaces the entire object server-side. Returns the updated object.
For partial updates, use L</patch> instead.
=head2 patch
my $patched = $api->patch('Pod', 'my-pod',
namespace => 'default',
patch => { metadata => { labels => { env => 'staging' } } },
);
# Or with an object:
my $patched = $api->patch($pod,
patch => { metadata => { labels => { env => 'staging' } } },
);
Partially update a resource. Unlike C<update()> which replaces the entire object, C<patch()> only modifies specified fields.
Supports three patch strategies via the C<type> parameter:
=over 4
=item C<strategic> (default) - Strategic Merge Patch (Kubernetes-native, understands array merge semantics)
=item C<merge> - JSON Merge Patch (RFC 7396, simple recursive merge)
=item C<json> - JSON Patch (RFC 6902, array of operations)
=back
lib/Kubernetes/REST.pm view on Meta::CPAN
$api->ensure_only(
label => 'app.kubernetes.io/component=queen',
objects => \@objects,
kinds => [qw(Role RoleBinding ClusterRoleBinding)],
namespaces => ['default', 'kube-system', undef],
);
Like L</ensure_all>, but also B<deletes> any resources matching the label
selector in the given kinds and namespaces that are not present in C<objects>.
Use this for resources where stale objects must not survive (e.g. RBAC).
Pass C<undef> inside C<namespaces> to scan cluster-scoped resources. If
C<namespaces> is omitted, only cluster-scoped resources are scanned.
Returns the list of applied objects (from L</ensure_all>).
=head2 watch
my $last_rv = $api->watch('Pod',
namespace => 'default',
on_event => sub {
my ($event) = @_;
say $event->type . ": " . $event->object->metadata->name;
},
timeout => 300,
resourceVersion => '12345',
labelSelector => 'app=web',
fieldSelector => 'status.phase=Running',
);
Watch for changes to resources. Uses the Kubernetes Watch API with chunked transfer encoding to stream events. The call blocks until the server-side timeout expires.
Returns the last C<resourceVersion> seen. Croaks on 410 Gone (resourceVersion too old).
See L<Kubernetes::REST/watch> for detailed documentation and resumable watch patterns.
=head2 log
# One-shot: get full log as string
my $text = $api->log('Pod', 'my-pod',
namespace => 'default',
tailLines => 100,
);
# Streaming: callback per log line
$api->log('Pod', 'my-pod',
namespace => 'default',
follow => 1,
on_line => sub {
my ($event) = @_; # Kubernetes::REST::LogEvent
say $event->line;
},
);
Retrieve logs from a pod. Supports two modes:
B<One-shot> (without C<on_line>): Returns the full log text as a string.
B<Streaming> (with C<on_line>): Calls the callback for each log line with a L<Kubernetes::REST::LogEvent> object. Blocks until the stream ends (or the server closes the connection).
The streaming mode is designed for event-based systems like L<IO::Async> — see L<Net::Async::Kubernetes> for async integration.
=head2 port_forward
my $session = $api->port_forward('Pod', 'my-pod',
namespace => 'default',
ports => [8080, 8443],
on_frame => sub { my ($channel, $payload) = @_; ... },
);
Start a full-duplex pod port-forward session.
This method requires an IO backend that implements C<call_duplex>. The default
L<Kubernetes::REST::LWPIO> and L<Kubernetes::REST::HTTPTinyIO> backends do not
currently provide duplex transport.
Returns whatever the IO backend returns for C<call_duplex> (typically a
session/handle object managed by that backend).
=head2 exec
my $session = $api->exec('Pod', 'my-pod',
namespace => 'default',
command => ['sh', '-c', 'echo hello'],
stdin => 0,
stdout => 1,
stderr => 1,
tty => 0,
on_frame => sub { my ($channel, $payload) = @_; ... },
);
Start a full-duplex pod exec session via the C</exec> subresource.
This method requires an IO backend that implements C<call_duplex>. The default
L<Kubernetes::REST::LWPIO> and L<Kubernetes::REST::HTTPTinyIO> backends do not
currently provide duplex transport.
Returns whatever the IO backend returns for C<call_duplex> (typically a
session/handle object managed by that backend).
=head2 attach
my $session = $api->attach('Pod', 'my-pod',
namespace => 'default',
container => 'app',
stdin => 1,
stdout => 1,
stderr => 1,
tty => 0,
on_frame => sub { my ($channel, $payload) = @_; ... },
);
Start a full-duplex pod attach session via the C</attach> subresource.
This method requires an IO backend that implements C<call_duplex>. The default
L<Kubernetes::REST::LWPIO> and L<Kubernetes::REST::HTTPTinyIO> backends do not
currently provide duplex transport.
Returns whatever the IO backend returns for C<call_duplex> (typically a
session/handle object managed by that backend).
lib/Kubernetes/REST.pm view on Meta::CPAN
=head1 UPGRADING FROM 0.02
B<WARNING: Version 1.00 contains breaking changes!>
This version has been completely rewritten. Key changes that may affect your code:
=over 4
=item * B<New simplified API>
The old method-per-operation API (e.g., C<< $api->Core->ListNamespacedPod(...) >>)
has been replaced with a simple API: C<list>, C<get>, C<create>, C<update>,
C<patch>, C<delete>, C<ensure>, C<ensure_all>, C<ensure_only>, C<watch>,
C<log>, C<port_forward>, C<exec>, C<attach>.
=item * B<Old API still works but deprecated>
The old API is still available for backwards compatibility but will emit deprecation
warnings. Set C<$ENV{HIDE_KUBERNETES_REST_V0_API_WARNING}> to suppress warnings.
=item * B<Uses IO::K8s classes>
Results are now returned as typed L<IO::K8s> objects instead of raw hashrefs.
Lists are returned as L<IO::K8s::List> objects.
B<Note:> L<IO::K8s> has also been completely rewritten (Moose to Moo, updated
to Kubernetes v1.31 API). See L<IO::K8s/"UPGRADING FROM 0.04"> for details.
=item * B<Short resource names>
You can now use short names like C<'Pod'> instead of full class paths. The
C<resource_map> attribute controls this mapping.
=item * B<Dynamic resource map>
Use C<resource_map_from_cluster =E<gt> 1> to load the resource map from the
cluster's OpenAPI spec, ensuring compatibility with any Kubernetes version.
=back
=head1 ATTRIBUTES
=head2 server
Required. Connection details for the Kubernetes API server. Can be a hashref or
a L<Kubernetes::REST::Server> object.
server => { endpoint => 'https://kubernetes.local:6443' }
=head2 credentials
Required. Authentication credentials. Can be a hashref or a L<Kubernetes::REST::AuthToken>
object.
credentials => { token => $bearer_token }
=head2 io
Optional. HTTP backend for making requests. Must consume the
L<Kubernetes::REST::Role::IO> role (i.e. implement C<call($req)> and
C<call_streaming($req, $callback)>; optional C<call_duplex($req, %callbacks)> for
full-duplex subresources such as pod port-forward). Defaults to L<Kubernetes::REST::LWPIO>
(L<LWP::UserAgent>), which supports L<LWP::ConsoleLogger> for HTTP debugging.
To use the lighter L<HTTP::Tiny> backend instead:
use Kubernetes::REST::HTTPTinyIO;
my $api = Kubernetes::REST->new(
server => ...,
credentials => ...,
io => Kubernetes::REST::HTTPTinyIO->new(
ssl_verify_server => 1,
),
);
To use an async event loop, provide your own IO backend:
my $api = Kubernetes::REST->new(
server => ...,
credentials => ...,
io => My::AsyncIO->new(loop => $loop),
);
=head2 k8s
Optional. L<IO::K8s> instance configured with the same resource map as this client.
Automatically created when needed.
=head2 resource_map_from_cluster
Optional boolean. If true, loads the resource map dynamically from the cluster's
OpenAPI spec. Defaults to true (loads from cluster).
resource_map_from_cluster => 1
=head2 resource_map
Optional hashref. Maps short resource names to IO::K8s class paths. By default
loads dynamically from the cluster (if C<resource_map_from_cluster> is true) or
uses L<IO::K8s> built-in map. Can be overridden for custom resources.
resource_map => { MyResource => 'Custom::V1::MyResource' }
=head2 cluster_version
Read-only. The Kubernetes cluster version string (e.g., "v1.31.0"). Fetched
automatically from the /version endpoint when first accessed.
=head1 METHODS
=head2 new_object($class, \%attrs) or new_object($class, %attrs)
Create a new IO::K8s object. Accepts short class names (e.g., 'Pod', 'Namespace')
and either a hashref or a hash of attributes.
# With hashref
my $ns = $api->new_object(Namespace => { metadata => { name => 'foo' } });
# With hash
my $ns = $api->new_object(Namespace => metadata => { name => 'foo' });
lib/Kubernetes/REST.pm view on Meta::CPAN
Returns a hashref mapping short resource names (e.g., "Pod") to full IO::K8s
class paths. This method is called automatically if C<resource_map_from_cluster>
is enabled.
=head1 BUILDING BLOCKS FOR ASYNC WRAPPERS
Async wrappers like L<Net::Async::Kubernetes> need access to the request/response
pipeline without going through the synchronous convenience methods. The following
public methods provide this:
=over 4
=item * C<expand_class($short)> - Resolve short name to full class
=item * C<build_path($class, %args)> - Build REST API URL path
=item * C<prepare_request($method, $path, %opts)> - Build HTTP request with auth
=item * C<check_response($response, $context)> - Validate HTTP status
=item * C<inflate_object($class, $response)> - JSON to typed object
=item * C<inflate_list($class, $response)> - JSON to typed list
=item * C<process_watch_chunk($class, \$buf, $chunk)> - Parse NDJSON watch stream
=item * C<process_log_chunk(\$buf, $chunk)> - Parse plain-text log stream
=back
Example async integration:
# Build request using Kubernetes::REST
my $class = $rest->expand_class('Pod');
my $path = $rest->build_path($class, name => $name, namespace => $ns) . '/log';
my $req = $rest->prepare_request('GET', $path, parameters => { follow => 'true' });
# Execute through your own event loop
my $buffer = '';
$async_http->request($req->url, sub {
my ($chunk) = @_;
for my $event ($rest->process_log_chunk(\$buffer, $chunk)) {
$on_line->($event);
}
});
=head1 PLUGGABLE IO ARCHITECTURE
The HTTP transport is decoupled from request preparation and response
processing. This makes it possible to swap the default L<LWP::UserAgent>
backend for L<HTTP::Tiny> or an async backend (e.g. L<Net::Async::HTTP>)
without changing any API logic.
The pipeline for each API call:
1. prepare_request() - builds HTTPRequest (method, url, headers, body)
2. io->call() - executes request (pluggable backend)
3. check_response() - validates HTTP status
4. inflate_object/list - decodes JSON + inflates IO::K8s objects
For watch, step 2 uses C<io-E<gt>call_streaming()> and step 4 uses
C<process_watch_chunk()> which parses NDJSON and inflates each event.
For log, step 2 uses C<io-E<gt>call_streaming()> and step 4 uses
C<process_log_chunk()> which parses plain-text lines into L<Kubernetes::REST::LogEvent> objects.
To implement a custom IO backend, consume L<Kubernetes::REST::Role::IO>
and implement C<call($req)> and C<call_streaming($req, $callback)>.
See L<Kubernetes::REST::LWPIO> and L<Kubernetes::REST::HTTPTinyIO> for
reference implementations.
=head1 SEE ALSO
=head2 Related Modules
=over
=item * L<IO::K8s> - Kubernetes resource classes (required dependency)
=item * L<Net::Async::Kubernetes> - Async Kubernetes client for L<IO::Async>
=back
=head2 Configuration and Authentication
=over
=item * L<Kubernetes::REST::Kubeconfig> - Load settings from kubeconfig
=item * L<Kubernetes::REST::Server> - Server connection configuration
=item * L<Kubernetes::REST::AuthToken> - Authentication credentials
=back
=head2 HTTP Backends
=over
=item * L<Kubernetes::REST::Role::IO> - IO interface role
=item * L<Kubernetes::REST::LWPIO> - LWP::UserAgent backend (default)
=item * L<Kubernetes::REST::HTTPTinyIO> - HTTP::Tiny backend
=item * L<LWP::ConsoleLogger> - HTTP debugging for LWPIO
=back
=head2 Data Objects
=over
=item * L<Kubernetes::REST::WatchEvent> - Watch event object
=item * L<Kubernetes::REST::LogEvent> - Log event object
=item * L<Kubernetes::REST::HTTPRequest> - HTTP request object
=item * L<Kubernetes::REST::HTTPResponse> - HTTP response object
=back
=head2 CLI Tools
=over
=item * L<Kubernetes::REST::CLI> - CLI base class
( run in 0.974 second using v1.01-cache-2.11-cpan-600a1bdf6e4 )