karmada-io
/
karmada
mirror of https://github.com/karmada-io/karmada.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2098 from carlory/karmadactl-log 

use kubectl logs to implement karmadactl logs
This commit is contained in:
karmada-bot 2022-07-13 10:22:36 +08:00 committed by GitHub
parent 156a02a452 4ee623c1e2
commit a70f1fb120
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 887 additions and 334 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

364
pkg/karmadactl/logs.go
View File

@ -2,43 +2,35 @@ package karmadactl
import (
"context"
"errors"
"fmt"
"io"
"os"
"regexp"
"sync"
"time"
"github.com/spf13/cobra"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/cli-runtime/pkg/genericclioptions"
"k8s.io/client-go/rest"
kubectllogs "k8s.io/kubectl/pkg/cmd/logs"
cmdutil "k8s.io/kubectl/pkg/cmd/util"
"k8s.io/kubectl/pkg/polymorphichelpers"
karmadaclientset "github.com/karmada-io/karmada/pkg/generated/clientset/versioned"
"github.com/karmada-io/karmada/pkg/karmadactl/options"
"github.com/karmada-io/karmada/pkg/util/gclient"
"github.com/karmada-io/karmada/pkg/util/lifted"
)
const (
defaultPodLogsTimeout = 20 * time.Second
logsUsageStr = "logs [-f] [-p] (POD | TYPE/NAME) [-c CONTAINER] (-C CLUSTER)"
logsUsageStr = "logs [-f] [-p] (POD | TYPE/NAME) [-c CONTAINER] (-C CLUSTER)"
)
var (
selectorTail int64 = 10
logsUsageErrStr = fmt.Sprintf("expected '%s'.\nPOD or TYPE/NAME is a required argument for the logs command", logsUsageStr)
logsUsageErrStr = fmt.Sprintf("expected '%s'.\nPOD or TYPE/NAME is a required argument for the logs command", logsUsageStr)
)
// NewCmdLogs new logs command.
func NewCmdLogs(karmadaConfig KarmadaConfig, parentCommand string) *cobra.Command {
ioStreams := genericclioptions.IOStreams{In: getIn, Out: getOut, ErrOut: getErr}
o := NewCommandLogsOptions(ioStreams, false)
streams := genericclioptions.IOStreams{In: getIn, Out: getOut, ErrOut: getErr}
o := &LogsOptions{
KubectlLogsOptions: kubectllogs.NewLogsOptions(streams, false),
}
cmd := &cobra.Command{
Use: logsUsageStr,
Short: "Print the logs for a container in a pod in a cluster",
@ -59,8 +51,9 @@ func NewCmdLogs(karmadaConfig KarmadaConfig, parentCommand string) *cobra.Comman
}
o.GlobalCommandOptions.AddFlags(cmd.Flags())
o.AddFlags(cmd)
o.KubectlLogsOptions.AddFlags(cmd)
cmd.Flags().StringVarP(&o.Namespace, "namespace", "n", o.Namespace, "If present, the namespace scope for this CLI request")
cmd.Flags().StringVarP(&o.Cluster, "cluster", "C", "", "Specify a member cluster")
return cmd
}
@ -92,333 +85,58 @@ func logsExample(parentCommand string) string {
return example
}
// CommandLogsOptions contains the input to the logs command.
type CommandLogsOptions struct {
// LogsOptions contains the input to the logs command.
type LogsOptions struct {
// global flags
options.GlobalCommandOptions
Cluster string
Namespace string
ResourceArg string
AllContainers bool
Options runtime.Object
Resources []string
ConsumeRequestFn func(rest.ResponseWrapper, io.Writer) error
// PodLogOptions
SinceTime string
Since time.Duration
Follow bool
Previous bool
Timestamps bool
IgnoreLogErrors bool
LimitBytes int64
Tail int64
Container string
InsecureSkipTLSVerifyBackend bool
// whether or not a container name was given via --container
ContainerNameSpecified bool
Selector string
MaxFollowConcurrency int
Prefix bool
Object runtime.Object
GetPodTimeout time.Duration
RESTClientGetter genericclioptions.RESTClientGetter
LogsForObject polymorphichelpers.LogsForObjectFunc
genericclioptions.IOStreams
TailSpecified bool
containerNameFromRefSpecRegexp *regexp.Regexp
f cmdutil.Factory
}
// NewCommandLogsOptions returns a LogsOptions.
func NewCommandLogsOptions(streams genericclioptions.IOStreams, allContainers bool) *CommandLogsOptions {
return &CommandLogsOptions{
IOStreams: streams,
AllContainers: allContainers,
Tail: -1,
MaxFollowConcurrency: 5,
containerNameFromRefSpecRegexp: regexp.MustCompile(`spec\.(?:initContainers|containers|ephemeralContainers){(.+)}`),
}
}
// AddFlags adds flags to the specified FlagSet.
func (o *CommandLogsOptions) AddFlags(cmd *cobra.Command) {
cmd.Flags().BoolVar(&o.AllContainers, "all-containers", o.AllContainers, "Get all containers' logs in the pod(s).")
cmd.Flags().BoolVarP(&o.Follow, "follow", "f", o.Follow, "Specify if the logs should be streamed.")
cmd.Flags().BoolVar(&o.Timestamps, "timestamps", o.Timestamps, "Include timestamps on each line in the log output")
cmd.Flags().Int64Var(&o.LimitBytes, "limit-bytes", o.LimitBytes, "Maximum bytes of logs to return. Defaults to no limit.")
cmd.Flags().BoolVarP(&o.Previous, "previous", "p", o.Previous, "If true, print the logs for the previous instance of the container in a pod if it exists.")
cmd.Flags().Int64Var(&o.Tail, "tail", o.Tail, "Lines of recent log file to display. Defaults to -1 with no selector, showing all log lines otherwise 10, if a selector is provided.")
cmd.Flags().BoolVar(&o.IgnoreLogErrors, "ignore-errors", o.IgnoreLogErrors, "If watching / following pod logs, allow for any errors that occur to be non-fatal")
cmd.Flags().StringVar(&o.SinceTime, "since-time", o.SinceTime, "Only return logs after a specific date (RFC3339). Defaults to all logs. Only one of since-time / since may be used.")
cmd.Flags().DurationVar(&o.Since, "since", o.Since, "Only return logs newer than a relative duration like 5s, 2m, or 3h. Defaults to all logs. Only one of since-time / since may be used.")
cmd.Flags().StringVarP(&o.Container, "container", "c", o.Container, "Print the logs of this container")
cmd.Flags().BoolVar(&o.InsecureSkipTLSVerifyBackend, "insecure-skip-tls-verify-backend", o.InsecureSkipTLSVerifyBackend,
"Skip verifying the identity of the kubelet that logs are requested from. In theory, an attacker could provide invalid log content back. You might want to use this if your kubelet serving certificates have expired.")
cmdutil.AddPodRunningTimeoutFlag(cmd, defaultPodLogsTimeout)
cmd.Flags().StringVarP(&o.Selector, "selector", "l", o.Selector, "Selector (label query) to filter on.")
cmd.Flags().IntVar(&o.MaxFollowConcurrency, "max-log-requests", o.MaxFollowConcurrency, "Specify maximum number of concurrent logs to follow when using by a selector. Defaults to 5.")
cmd.Flags().BoolVar(&o.Prefix, "prefix", o.Prefix, "Prefix each log line with the log source (pod name and container name)")
cmd.Flags().StringVarP(&o.Namespace, "namespace", "n", "default", "-n=namespace or -n namespace")
cmd.Flags().StringVarP(&o.Cluster, "cluster", "C", "", "Specify a member cluster")
// flags specific to logs
KubectlLogsOptions *kubectllogs.LogsOptions
Namespace string
Cluster string
}
// Complete ensures that options are valid and marshals them if necessary
func (o *CommandLogsOptions) Complete(karmadaConfig KarmadaConfig, cmd *cobra.Command, args []string) error {
o.ContainerNameSpecified = cmd.Flag("container").Changed
o.TailSpecified = cmd.Flag("tail").Changed
o.Resources = args
func (o *LogsOptions) Complete(karmadaConfig KarmadaConfig, cmd *cobra.Command, args []string) error {
if o.Cluster == "" {
return fmt.Errorf("must specify a cluster")
}
// print correct usage message when the given arguments are invalid
switch len(args) {
case 0:
if len(o.Selector) == 0 {
if len(o.KubectlLogsOptions.Selector) == 0 {
return cmdutil.UsageErrorf(cmd, "%s", logsUsageErrStr)
}
case 1:
o.ResourceArg = args[0]
if len(o.Selector) != 0 {
if len(o.KubectlLogsOptions.Selector) != 0 {
return cmdutil.UsageErrorf(cmd, "only a selector (-l) or a POD name is allowed")
}
case 2:
o.ResourceArg = args[0]
o.Container = args[1]
default:
return cmdutil.UsageErrorf(cmd, "%s", logsUsageErrStr)
}
var err error
o.ConsumeRequestFn = lifted.DefaultConsumeRequest
o.GetPodTimeout, err = cmdutil.GetPodRunningTimeoutFlag(cmd)
if err != nil {
return err
}
o.Options, err = o.toLogOptions()
if err != nil {
return err
}
if len(o.Cluster) == 0 {
return fmt.Errorf("must specify a cluster")
}
karmadaRestConfig, err := karmadaConfig.GetRestConfig(o.KarmadaContext, o.KubeConfig)
if err != nil {
return fmt.Errorf("failed to get control plane rest config. context: %s, kube-config: %s, error: %v",
o.KarmadaContext, o.KubeConfig, err)
}
clusterInfo, err := getClusterInfo(karmadaRestConfig, o.Cluster, o.KubeConfig, o.KarmadaContext)
if err != nil {
return err
}
o.f = getFactory(o.Cluster, clusterInfo, "")
o.RESTClientGetter = o.f
o.LogsForObject = polymorphichelpers.LogsForObjectFn
if err := o.completeObj(); err != nil {
return err
}
return nil
f := getFactory(o.Cluster, clusterInfo, o.Namespace)
return o.KubectlLogsOptions.Complete(f, cmd, args)
}
// Validate checks to the LogsOptions to see if there is sufficient information run the command
func (o *CommandLogsOptions) Validate() error {
if len(o.SinceTime) > 0 && o.Since != 0 {
return fmt.Errorf("at most one of `sinceTime` or `sinceSeconds` may be specified")
}
logsOptions, ok := o.Options.(*corev1.PodLogOptions)
if !ok {
return errors.New("unexpected logs options object")
}
if o.AllContainers && len(logsOptions.Container) > 0 {
return fmt.Errorf("--all-containers=true should not be specified with container name %s", logsOptions.Container)
}
if o.ContainerNameSpecified && len(o.Resources) == 2 {
return fmt.Errorf("only one of -c or an inline [CONTAINER] arg is allowed")
}
if o.LimitBytes < 0 {
return fmt.Errorf("--limit-bytes must be greater than 0")
}
if logsOptions.SinceSeconds != nil && *logsOptions.SinceSeconds < int64(0) {
return fmt.Errorf("--since must be greater than 0")
}
if logsOptions.TailLines != nil && *logsOptions.TailLines < -1 {
return fmt.Errorf("--tail must be greater than or equal to -1")
}
return nil
func (o *LogsOptions) Validate() error {
return o.KubectlLogsOptions.Validate()
}
// Run retrieves a pod log
func (o *CommandLogsOptions) Run() error {
requests, err := o.LogsForObject(o.RESTClientGetter, o.Object, o.Options, o.GetPodTimeout, o.AllContainers)
if err != nil {
return err
}
if o.Follow && len(requests) > 1 {
if len(requests) > o.MaxFollowConcurrency {
return fmt.Errorf(
"you are attempting to follow %d log streams, but maximum allowed concurrency is %d, use --max-log-requests to increase the limit",
len(requests), o.MaxFollowConcurrency,
)
}
return o.parallelConsumeRequest(requests)
}
return o.sequentialConsumeRequest(requests)
}
func (o *CommandLogsOptions) parallelConsumeRequest(requests map[corev1.ObjectReference]rest.ResponseWrapper) error {
reader, writer := io.Pipe()
wg := &sync.WaitGroup{}
wg.Add(len(requests))
for objRef, request := range requests {
go func(objRef corev1.ObjectReference, request rest.ResponseWrapper) {
defer wg.Done()
out := o.addPrefixIfNeeded(objRef, writer)
if err := o.ConsumeRequestFn(request, out); err != nil {
if !o.IgnoreLogErrors {
_ = writer.CloseWithError(err)
// It's important to return here to propagate the error via the pipe
return
}
fmt.Fprintf(writer, "error: %v\n", err)
}
}(objRef, request)
}
go func() {
wg.Wait()
writer.Close()
}()
_, err := io.Copy(o.Out, reader)
return err
}
func (o *CommandLogsOptions) completeObj() error {
if o.Object == nil {
builder := o.f.NewBuilder().
WithScheme(gclient.NewSchema(), gclient.NewSchema().PrioritizedVersionsAllGroups()...).
NamespaceParam(o.Namespace).DefaultNamespace().
SingleResourceType()
if o.ResourceArg != "" {
builder.ResourceNames("pods", o.ResourceArg)
}
if o.Selector != "" {
builder.ResourceTypes("pods").LabelSelectorParam(o.Selector)
}
infos, err := builder.Do().Infos()
if err != nil {
return err
}
if o.Selector == "" && len(infos) != 1 {
return errors.New("expected a resource")
}
o.Object = infos[0].Object
if o.Selector != "" && len(o.Object.(*corev1.PodList).Items) == 0 {
fmt.Fprintf(o.ErrOut, "No resources found in %s namespace.\n", o.Namespace)
}
}
return nil
}
func (o *CommandLogsOptions) toLogOptions() (*corev1.PodLogOptions, error) {
logOptions := &corev1.PodLogOptions{
Container: o.Container,
Follow: o.Follow,
Previous: o.Previous,
Timestamps: o.Timestamps,
InsecureSkipTLSVerifyBackend: o.InsecureSkipTLSVerifyBackend,
}
if len(o.SinceTime) > 0 {
t, err := lifted.ParseRFC3339(o.SinceTime, metav1.Now)
if err != nil {
return nil, err
}
logOptions.SinceTime = &t
}
if o.LimitBytes != 0 {
logOptions.LimitBytes = &o.LimitBytes
}
if o.Since != 0 {
// round up to the nearest second
sec := int64(o.Since.Round(time.Second))
logOptions.SinceSeconds = &sec
}
if len(o.Selector) > 0 && o.Tail == -1 && !o.TailSpecified {
logOptions.TailLines = &selectorTail
} else if o.Tail != -1 {
logOptions.TailLines = &o.Tail
}
return logOptions, nil
}
func (o *CommandLogsOptions) sequentialConsumeRequest(requests map[corev1.ObjectReference]rest.ResponseWrapper) error {
for objRef, request := range requests {
out := o.addPrefixIfNeeded(objRef, o.Out)
if err := o.ConsumeRequestFn(request, out); err != nil {
if !o.IgnoreLogErrors {
return err
}
fmt.Fprintf(o.Out, "error: %v\n", err)
}
}
return nil
}
func (o *CommandLogsOptions) addPrefixIfNeeded(ref corev1.ObjectReference, writer io.Writer) io.Writer {
if !o.Prefix || ref.FieldPath == "" || ref.Name == "" {
return writer
}
// We rely on ref.FieldPath to contain a reference to a container
// including a container name (not an index) so we can get a container name
// without making an extra API request.
var containerName string
containerNameMatches := o.containerNameFromRefSpecRegexp.FindStringSubmatch(ref.FieldPath)
if len(containerNameMatches) == 2 {
containerName = containerNameMatches[1]
}
prefix := fmt.Sprintf("[pod/%s/%s] ", ref.Name, containerName)
return &prefixingWriter{
prefix: []byte(prefix),
writer: writer,
}
func (o *LogsOptions) Run() error {
return o.KubectlLogsOptions.RunLogs()
}
// getClusterInfo get information of cluster
@ -448,25 +166,3 @@ func getClusterInfo(karmadaRestConfig *rest.Config, clusterName, kubeConfig, kar
return clusterInfos, nil
}
type prefixingWriter struct {
prefix []byte
writer io.Writer
}
func (pw *prefixingWriter) Write(p []byte) (int, error) {
if len(p) == 0 {
return 0, nil
}
// Perform an "atomic" write of a prefix and p to make sure that it doesn't interleave
// sub-line when used concurrently with io.PipeWrite.
n, err := pw.writer.Write(append(pw.prefix, p...))
if n > len(p) {
// To comply with the io.Writer interface requirements we must
// return a number of bytes written from p (0 <= n <= len(p)),
// so we are ignoring the length of the prefix here.
return len(p), err
}
return n, err
}

463
vendor/k8s.io/kubectl/pkg/cmd/logs/logs.go generated vendored Normal file
View File

@ -0,0 +1,463 @@
/*
Copyright 2014 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package logs
import (
"bufio"
"context"
"errors"
"fmt"
"io"
"regexp"
"sync"
"time"
"github.com/spf13/cobra"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/cli-runtime/pkg/genericclioptions"
"k8s.io/client-go/rest"
cmdutil "k8s.io/kubectl/pkg/cmd/util"
"k8s.io/kubectl/pkg/polymorphichelpers"
"k8s.io/kubectl/pkg/scheme"
"k8s.io/kubectl/pkg/util"
"k8s.io/kubectl/pkg/util/completion"
"k8s.io/kubectl/pkg/util/i18n"
"k8s.io/kubectl/pkg/util/templates"
)
const (
logsUsageStr = "logs [-f] [-p] (POD | TYPE/NAME) [-c CONTAINER]"
)
var (
logsLong = templates.LongDesc(i18n.T(`
Print the logs for a container in a pod or specified resource.
If the pod has only one container, the container name is optional.`))
logsExample = templates.Examples(i18n.T(`
# Return snapshot logs from pod nginx with only one container
kubectl logs nginx
# Return snapshot logs from pod nginx with multi containers
kubectl logs nginx --all-containers=true
# Return snapshot logs from all containers in pods defined by label app=nginx
kubectl logs -l app=nginx --all-containers=true
# Return snapshot of previous terminated ruby container logs from pod web-1
kubectl logs -p -c ruby web-1
# Begin streaming the logs of the ruby container in pod web-1
kubectl logs -f -c ruby web-1
# Begin streaming the logs from all containers in pods defined by label app=nginx
kubectl logs -f -l app=nginx --all-containers=true
# Display only the most recent 20 lines of output in pod nginx
kubectl logs --tail=20 nginx
# Show all logs from pod nginx written in the last hour
kubectl logs --since=1h nginx
# Show logs from a kubelet with an expired serving certificate
kubectl logs --insecure-skip-tls-verify-backend nginx
# Return snapshot logs from first container of a job named hello
kubectl logs job/hello
# Return snapshot logs from container nginx-1 of a deployment named nginx
kubectl logs deployment/nginx -c nginx-1`))
selectorTail int64 = 10
logsUsageErrStr = fmt.Sprintf("expected '%s'.\nPOD or TYPE/NAME is a required argument for the logs command", logsUsageStr)
)
const (
defaultPodLogsTimeout = 20 * time.Second
)
type LogsOptions struct {
Namespace string
ResourceArg string
AllContainers bool
Options runtime.Object
Resources []string
ConsumeRequestFn func(rest.ResponseWrapper, io.Writer) error
// PodLogOptions
SinceTime string
SinceSeconds time.Duration
Follow bool
Previous bool
Timestamps bool
IgnoreLogErrors bool
LimitBytes int64
Tail int64
Container string
InsecureSkipTLSVerifyBackend bool
// whether or not a container name was given via --container
ContainerNameSpecified bool
Selector string
MaxFollowConcurrency int
Prefix bool
Object runtime.Object
GetPodTimeout time.Duration
RESTClientGetter genericclioptions.RESTClientGetter
LogsForObject polymorphichelpers.LogsForObjectFunc
genericclioptions.IOStreams
TailSpecified bool
containerNameFromRefSpecRegexp *regexp.Regexp
}
func NewLogsOptions(streams genericclioptions.IOStreams, allContainers bool) *LogsOptions {
return &LogsOptions{
IOStreams: streams,
AllContainers: allContainers,
Tail: -1,
MaxFollowConcurrency: 5,
containerNameFromRefSpecRegexp: regexp.MustCompile(`spec\.(?:initContainers|containers|ephemeralContainers){(.+)}`),
}
}
// NewCmdLogs creates a new pod logs command
func NewCmdLogs(f cmdutil.Factory, streams genericclioptions.IOStreams) *cobra.Command {
o := NewLogsOptions(streams, false)
cmd := &cobra.Command{
Use: logsUsageStr,
DisableFlagsInUseLine: true,
Short: i18n.T("Print the logs for a container in a pod"),
Long: logsLong,
Example: logsExample,
ValidArgsFunction: completion.PodResourceNameAndContainerCompletionFunc(f),
Run: func(cmd *cobra.Command, args []string) {
cmdutil.CheckErr(o.Complete(f, cmd, args))
cmdutil.CheckErr(o.Validate())
cmdutil.CheckErr(o.RunLogs())
},
}
o.AddFlags(cmd)
return cmd
}
func (o *LogsOptions) AddFlags(cmd *cobra.Command) {
cmd.Flags().BoolVar(&o.AllContainers, "all-containers", o.AllContainers, "Get all containers' logs in the pod(s).")
cmd.Flags().BoolVarP(&o.Follow, "follow", "f", o.Follow, "Specify if the logs should be streamed.")
cmd.Flags().BoolVar(&o.Timestamps, "timestamps", o.Timestamps, "Include timestamps on each line in the log output")
cmd.Flags().Int64Var(&o.LimitBytes, "limit-bytes", o.LimitBytes, "Maximum bytes of logs to return. Defaults to no limit.")
cmd.Flags().BoolVarP(&o.Previous, "previous", "p", o.Previous, "If true, print the logs for the previous instance of the container in a pod if it exists.")
cmd.Flags().Int64Var(&o.Tail, "tail", o.Tail, "Lines of recent log file to display. Defaults to -1 with no selector, showing all log lines otherwise 10, if a selector is provided.")
cmd.Flags().BoolVar(&o.IgnoreLogErrors, "ignore-errors", o.IgnoreLogErrors, "If watching / following pod logs, allow for any errors that occur to be non-fatal")
cmd.Flags().StringVar(&o.SinceTime, "since-time", o.SinceTime, i18n.T("Only return logs after a specific date (RFC3339). Defaults to all logs. Only one of since-time / since may be used."))
cmd.Flags().DurationVar(&o.SinceSeconds, "since", o.SinceSeconds, "Only return logs newer than a relative duration like 5s, 2m, or 3h. Defaults to all logs. Only one of since-time / since may be used.")
cmd.Flags().StringVarP(&o.Container, "container", "c", o.Container, "Print the logs of this container")
cmd.Flags().BoolVar(&o.InsecureSkipTLSVerifyBackend, "insecure-skip-tls-verify-backend", o.InsecureSkipTLSVerifyBackend,
"Skip verifying the identity of the kubelet that logs are requested from. In theory, an attacker could provide invalid log content back. You might want to use this if your kubelet serving certificates have expired.")
cmdutil.AddPodRunningTimeoutFlag(cmd, defaultPodLogsTimeout)
cmdutil.AddLabelSelectorFlagVar(cmd, &o.Selector)
cmd.Flags().IntVar(&o.MaxFollowConcurrency, "max-log-requests", o.MaxFollowConcurrency, "Specify maximum number of concurrent logs to follow when using by a selector. Defaults to 5.")
cmd.Flags().BoolVar(&o.Prefix, "prefix", o.Prefix, "Prefix each log line with the log source (pod name and container name)")
}
func (o *LogsOptions) ToLogOptions() (*corev1.PodLogOptions, error) {
logOptions := &corev1.PodLogOptions{
Container: o.Container,
Follow: o.Follow,
Previous: o.Previous,
Timestamps: o.Timestamps,
InsecureSkipTLSVerifyBackend: o.InsecureSkipTLSVerifyBackend,
}
if len(o.SinceTime) > 0 {
t, err := util.ParseRFC3339(o.SinceTime, metav1.Now)
if err != nil {
return nil, err
}
logOptions.SinceTime = &t
}
if o.LimitBytes != 0 {
logOptions.LimitBytes = &o.LimitBytes
}
if o.SinceSeconds != 0 {
// round up to the nearest second
sec := int64(o.SinceSeconds.Round(time.Second).Seconds())
logOptions.SinceSeconds = &sec
}
if len(o.Selector) > 0 && o.Tail == -1 && !o.TailSpecified {
logOptions.TailLines = &selectorTail
} else if o.Tail != -1 {
logOptions.TailLines = &o.Tail
}
return logOptions, nil
}
func (o *LogsOptions) Complete(f cmdutil.Factory, cmd *cobra.Command, args []string) error {
o.ContainerNameSpecified = cmd.Flag("container").Changed
o.TailSpecified = cmd.Flag("tail").Changed
o.Resources = args
switch len(args) {
case 0:
if len(o.Selector) == 0 {
return cmdutil.UsageErrorf(cmd, "%s", logsUsageErrStr)
}
case 1:
o.ResourceArg = args[0]
if len(o.Selector) != 0 {
return cmdutil.UsageErrorf(cmd, "only a selector (-l) or a POD name is allowed")
}
case 2:
o.ResourceArg = args[0]
o.Container = args[1]
default:
return cmdutil.UsageErrorf(cmd, "%s", logsUsageErrStr)
}
var err error
o.Namespace, _, err = f.ToRawKubeConfigLoader().Namespace()
if err != nil {
return err
}
o.ConsumeRequestFn = DefaultConsumeRequest
o.GetPodTimeout, err = cmdutil.GetPodRunningTimeoutFlag(cmd)
if err != nil {
return err
}
o.Options, err = o.ToLogOptions()
if err != nil {
return err
}
o.RESTClientGetter = f
o.LogsForObject = polymorphichelpers.LogsForObjectFn
if o.Object == nil {
builder := f.NewBuilder().
WithScheme(scheme.Scheme, scheme.Scheme.PrioritizedVersionsAllGroups()...).
NamespaceParam(o.Namespace).DefaultNamespace().
SingleResourceType()
if o.ResourceArg != "" {
builder.ResourceNames("pods", o.ResourceArg)
}
if o.Selector != "" {
builder.ResourceTypes("pods").LabelSelectorParam(o.Selector)
}
infos, err := builder.Do().Infos()
if err != nil {
return err
}
if o.Selector == "" && len(infos) != 1 {
return errors.New("expected a resource")
}
o.Object = infos[0].Object
if o.Selector != "" && len(o.Object.(*corev1.PodList).Items) == 0 {
fmt.Fprintf(o.ErrOut, "No resources found in %s namespace.\n", o.Namespace)
}
}
return nil
}
func (o LogsOptions) Validate() error {
if len(o.SinceTime) > 0 && o.SinceSeconds != 0 {
return fmt.Errorf("at most one of `sinceTime` or `sinceSeconds` may be specified")
}
logsOptions, ok := o.Options.(*corev1.PodLogOptions)
if !ok {
return errors.New("unexpected logs options object")
}
if o.AllContainers && len(logsOptions.Container) > 0 {
return fmt.Errorf("--all-containers=true should not be specified with container name %s", logsOptions.Container)
}
if o.ContainerNameSpecified && len(o.Resources) == 2 {
return fmt.Errorf("only one of -c or an inline [CONTAINER] arg is allowed")
}
if o.LimitBytes < 0 {
return fmt.Errorf("--limit-bytes must be greater than 0")
}
if logsOptions.SinceSeconds != nil && *logsOptions.SinceSeconds < int64(0) {
return fmt.Errorf("--since must be greater than 0")
}
if logsOptions.TailLines != nil && *logsOptions.TailLines < -1 {
return fmt.Errorf("--tail must be greater than or equal to -1")
}
return nil
}
// RunLogs retrieves a pod log
func (o LogsOptions) RunLogs() error {
requests, err := o.LogsForObject(o.RESTClientGetter, o.Object, o.Options, o.GetPodTimeout, o.AllContainers)
if err != nil {
return err
}
if o.Follow && len(requests) > 1 {
if len(requests) > o.MaxFollowConcurrency {
return fmt.Errorf(
"you are attempting to follow %d log streams, but maximum allowed concurrency is %d, use --max-log-requests to increase the limit",
len(requests), o.MaxFollowConcurrency,
)
}
return o.parallelConsumeRequest(requests)
}
return o.sequentialConsumeRequest(requests)
}
func (o LogsOptions) parallelConsumeRequest(requests map[corev1.ObjectReference]rest.ResponseWrapper) error {
reader, writer := io.Pipe()
wg := &sync.WaitGroup{}
wg.Add(len(requests))
for objRef, request := range requests {
go func(objRef corev1.ObjectReference, request rest.ResponseWrapper) {
defer wg.Done()
out := o.addPrefixIfNeeded(objRef, writer)
if err := o.ConsumeRequestFn(request, out); err != nil {
if !o.IgnoreLogErrors {
writer.CloseWithError(err)
// It's important to return here to propagate the error via the pipe
return
}
fmt.Fprintf(writer, "error: %v\n", err)
}
}(objRef, request)
}
go func() {
wg.Wait()
writer.Close()
}()
_, err := io.Copy(o.Out, reader)
return err
}
func (o LogsOptions) sequentialConsumeRequest(requests map[corev1.ObjectReference]rest.ResponseWrapper) error {
for objRef, request := range requests {
out := o.addPrefixIfNeeded(objRef, o.Out)
if err := o.ConsumeRequestFn(request, out); err != nil {
if !o.IgnoreLogErrors {
return err
}
fmt.Fprintf(o.Out, "error: %v\n", err)
}
}
return nil
}
func (o LogsOptions) addPrefixIfNeeded(ref corev1.ObjectReference, writer io.Writer) io.Writer {
if !o.Prefix || ref.FieldPath == "" || ref.Name == "" {
return writer
}
// We rely on ref.FieldPath to contain a reference to a container
// including a container name (not an index) so we can get a container name
// without making an extra API request.
var containerName string
containerNameMatches := o.containerNameFromRefSpecRegexp.FindStringSubmatch(ref.FieldPath)
if len(containerNameMatches) == 2 {
containerName = containerNameMatches[1]
}
prefix := fmt.Sprintf("[pod/%s/%s] ", ref.Name, containerName)
return &prefixingWriter{
prefix: []byte(prefix),
writer: writer,
}
}
// DefaultConsumeRequest reads the data from request and writes into
// the out writer. It buffers data from requests until the newline or io.EOF
// occurs in the data, so it doesn't interleave logs sub-line
// when running concurrently.
//
// A successful read returns err == nil, not err == io.EOF.
// Because the function is defined to read from request until io.EOF, it does
// not treat an io.EOF as an error to be reported.
func DefaultConsumeRequest(request rest.ResponseWrapper, out io.Writer) error {
readCloser, err := request.Stream(context.TODO())
if err != nil {
return err
}
defer readCloser.Close()
r := bufio.NewReader(readCloser)
for {
bytes, err := r.ReadBytes('\n')
if _, err := out.Write(bytes); err != nil {
return err
}
if err != nil {
if err != io.EOF {
return err
}
return nil
}
}
}
type prefixingWriter struct {
prefix []byte
writer io.Writer
}
func (pw *prefixingWriter) Write(p []byte) (int, error) {
if len(p) == 0 {
return 0, nil
}
// Perform an "atomic" write of a prefix and p to make sure that it doesn't interleave
// sub-line when used concurrently with io.PipeWrite.
n, err := pw.writer.Write(append(pw.prefix, p...))
if n > len(p) {
// To comply with the io.Writer interface requirements we must
// return a number of bytes written from p (0 <= n <= len(p)),
// so we are ignoring the length of the prefix here.
return len(p), err
}
return n, err
}

146
vendor/k8s.io/kubectl/pkg/util/apply.go generated vendored Normal file
View File

@ -0,0 +1,146 @@
/*
Copyright 2014 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package util
import (
"k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/meta"
"k8s.io/apimachinery/pkg/runtime"
)
var metadataAccessor = meta.NewAccessor()
// GetOriginalConfiguration retrieves the original configuration of the object
// from the annotation, or nil if no annotation was found.
func GetOriginalConfiguration(obj runtime.Object) ([]byte, error) {
annots, err := metadataAccessor.Annotations(obj)
if err != nil {
return nil, err
}
if annots == nil {
return nil, nil
}
original, ok := annots[v1.LastAppliedConfigAnnotation]
if !ok {
return nil, nil
}
return []byte(original), nil
}
// SetOriginalConfiguration sets the original configuration of the object
// as the annotation on the object for later use in computing a three way patch.
func setOriginalConfiguration(obj runtime.Object, original []byte) error {
if len(original) < 1 {
return nil
}
annots, err := metadataAccessor.Annotations(obj)
if err != nil {
return err
}
if annots == nil {
annots = map[string]string{}
}
annots[v1.LastAppliedConfigAnnotation] = string(original)
return metadataAccessor.SetAnnotations(obj, annots)
}
// GetModifiedConfiguration retrieves the modified configuration of the object.
// If annotate is true, it embeds the result as an annotation in the modified
// configuration. If an object was read from the command input, it will use that
// version of the object. Otherwise, it will use the version from the server.
func GetModifiedConfiguration(obj runtime.Object, annotate bool, codec runtime.Encoder) ([]byte, error) {
// First serialize the object without the annotation to prevent recursion,
// then add that serialization to it as the annotation and serialize it again.
var modified []byte
// Otherwise, use the server side version of the object.
// Get the current annotations from the object.
annots, err := metadataAccessor.Annotations(obj)
if err != nil {
return nil, err
}
if annots == nil {
annots = map[string]string{}
}
original := annots[v1.LastAppliedConfigAnnotation]
delete(annots, v1.LastAppliedConfigAnnotation)
if err := metadataAccessor.SetAnnotations(obj, annots); err != nil {
return nil, err
}
modified, err = runtime.Encode(codec, obj)
if err != nil {
return nil, err
}
if annotate {
annots[v1.LastAppliedConfigAnnotation] = string(modified)
if err := metadataAccessor.SetAnnotations(obj, annots); err != nil {
return nil, err
}
modified, err = runtime.Encode(codec, obj)
if err != nil {
return nil, err
}
}
// Restore the object to its original condition.
annots[v1.LastAppliedConfigAnnotation] = original
if err := metadataAccessor.SetAnnotations(obj, annots); err != nil {
return nil, err
}
return modified, nil
}
// updateApplyAnnotation calls CreateApplyAnnotation if the last applied
// configuration annotation is already present. Otherwise, it does nothing.
func updateApplyAnnotation(obj runtime.Object, codec runtime.Encoder) error {
if original, err := GetOriginalConfiguration(obj); err != nil || len(original) <= 0 {
return err
}
return CreateApplyAnnotation(obj, codec)
}
// CreateApplyAnnotation gets the modified configuration of the object,
// without embedding it again, and then sets it on the object as the annotation.
func CreateApplyAnnotation(obj runtime.Object, codec runtime.Encoder) error {
modified, err := GetModifiedConfiguration(obj, false, codec)
if err != nil {
return err
}
return setOriginalConfiguration(obj, modified)
}
// CreateOrUpdateAnnotation creates the annotation used by
// kubectl apply only when createAnnotation is true
// Otherwise, only update the annotation when it already exists
func CreateOrUpdateAnnotation(createAnnotation bool, obj runtime.Object, codec runtime.Encoder) error {
if createAnnotation {
return CreateApplyAnnotation(obj, codec)
}
return updateApplyAnnotation(obj, codec)
}

36
vendor/k8s.io/kubectl/pkg/util/pod_port.go generated vendored Normal file
View File

@ -0,0 +1,36 @@
/*
Copyright 2018 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package util
import (
"fmt"
"k8s.io/api/core/v1"
)
// LookupContainerPortNumberByName find containerPort number by its named port name
func LookupContainerPortNumberByName(pod v1.Pod, name string) (int32, error) {
for _, ctr := range pod.Spec.Containers {
for _, ctrportspec := range ctr.Ports {
if ctrportspec.Name == name {
return ctrportspec.ContainerPort, nil
}
}
}
return int32(-1), fmt.Errorf("Pod '%s' does not have a named port '%s'", pod.Name, name)
}

59
vendor/k8s.io/kubectl/pkg/util/service_port.go generated vendored Normal file
View File

@ -0,0 +1,59 @@
/*
Copyright 2018 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package util
import (
"fmt"
"k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/util/intstr"
)
// LookupContainerPortNumberByServicePort implements
// the handling of resolving container named port, as well as ignoring targetPort when clusterIP=None
// It returns an error when a named port can't find a match (with -1 returned), or when the service does not
// declare such port (with the input port number returned).
func LookupContainerPortNumberByServicePort(svc v1.Service, pod v1.Pod, port int32) (int32, error) {
for _, svcportspec := range svc.Spec.Ports {
if svcportspec.Port != port {
continue
}
if svc.Spec.ClusterIP == v1.ClusterIPNone {
return port, nil
}
if svcportspec.TargetPort.Type == intstr.Int {
if svcportspec.TargetPort.IntValue() == 0 {
// targetPort is omitted, and the IntValue() would be zero
return svcportspec.Port, nil
}
return int32(svcportspec.TargetPort.IntValue()), nil
}
return LookupContainerPortNumberByName(pod, svcportspec.TargetPort.String())
}
return port, fmt.Errorf("Service %s does not have a service port %d", svc.Name, port)
}
// LookupServicePortNumberByName find service port number by its named port name
func LookupServicePortNumberByName(svc v1.Service, name string) (int32, error) {
for _, svcportspec := range svc.Spec.Ports {
if svcportspec.Name == name {
return svcportspec.Port, nil
}
}
return int32(-1), fmt.Errorf("Service '%s' does not have a named port '%s'", svc.Name, name)
}

29
vendor/k8s.io/kubectl/pkg/util/umask.go generated vendored Normal file
View File

@ -0,0 +1,29 @@
//go:build !windows
// +build !windows
/*
Copyright 2014 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package util
import (
"golang.org/x/sys/unix"
)
// Umask is a wrapper for `unix.Umask()` on non-Windows platforms
func Umask(mask int) (old int, err error) {
return unix.Umask(mask), nil
}

29
vendor/k8s.io/kubectl/pkg/util/umask_windows.go generated vendored Normal file
View File

@ -0,0 +1,29 @@
//go:build windows
// +build windows
/*
Copyright 2014 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package util
import (
"errors"
)
// Umask returns an error on Windows
func Umask(mask int) (int, error) {
return 0, errors.New("platform and architecture is not supported")
}

93
vendor/k8s.io/kubectl/pkg/util/util.go generated vendored Normal file
View File

@ -0,0 +1,93 @@
/*
Copyright 2017 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package util
import (
"crypto/md5"
"errors"
"fmt"
"path"
"path/filepath"
"strings"
"time"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
)
// ParseRFC3339 parses an RFC3339 date in either RFC3339Nano or RFC3339 format.
func ParseRFC3339(s string, nowFn func() metav1.Time) (metav1.Time, error) {
if t, timeErr := time.Parse(time.RFC3339Nano, s); timeErr == nil {
return metav1.Time{Time: t}, nil
}
t, err := time.Parse(time.RFC3339, s)
if err != nil {
return metav1.Time{}, err
}
return metav1.Time{Time: t}, nil
}
// HashObject returns the hash of a Object hash by a Codec
func HashObject(obj runtime.Object, codec runtime.Codec) (string, error) {
data, err := runtime.Encode(codec, obj)
if err != nil {
return "", err
}
return fmt.Sprintf("%x", md5.Sum(data)), nil
}
// ParseFileSource parses the source given.
//
// Acceptable formats include:
// 1. source-path: the basename will become the key name
// 2. source-name=source-path: the source-name will become the key name and
// source-path is the path to the key file.
//
// Key names cannot include '='.
func ParseFileSource(source string) (keyName, filePath string, err error) {
numSeparators := strings.Count(source, "=")
switch {
case numSeparators == 0:
return path.Base(filepath.ToSlash(source)), source, nil
case numSeparators == 1 && strings.HasPrefix(source, "="):
return "", "", fmt.Errorf("key name for file path %v missing", strings.TrimPrefix(source, "="))
case numSeparators == 1 && strings.HasSuffix(source, "="):
return "", "", fmt.Errorf("file path for key name %v missing", strings.TrimSuffix(source, "="))
case numSeparators > 1:
return "", "", errors.New("key names or file paths cannot contain '='")
default:
components := strings.Split(source, "=")
return components[0], components[1], nil
}
}
// ParseLiteralSource parses the source key=val pair into its component pieces.
// This functionality is distinguished from strings.SplitN(source, "=", 2) since
// it returns an error in the case of empty keys, values, or a missing equals sign.
func ParseLiteralSource(source string) (keyName, value string, err error) {
// leading equal is invalid
if strings.Index(source, "=") == 0 {
return "", "", fmt.Errorf("invalid literal source %v, expected key=value", source)
}
// split after the first equal (so values can have the = character)
items := strings.SplitN(source, "=", 2)
if len(items) != 2 {
return "", "", fmt.Errorf("invalid literal source %v, expected key=value", source)
}
return items[0], items[1], nil
}

2
vendor/modules.txt vendored
View File

@ -1410,12 +1410,14 @@ k8s.io/kubectl/pkg/apps
k8s.io/kubectl/pkg/cmd/apiresources
k8s.io/kubectl/pkg/cmd/exec
k8s.io/kubectl/pkg/cmd/get
k8s.io/kubectl/pkg/cmd/logs
k8s.io/kubectl/pkg/cmd/util
k8s.io/kubectl/pkg/cmd/util/podcmd
k8s.io/kubectl/pkg/describe
k8s.io/kubectl/pkg/polymorphichelpers
k8s.io/kubectl/pkg/rawhttp
k8s.io/kubectl/pkg/scheme
k8s.io/kubectl/pkg/util
k8s.io/kubectl/pkg/util/certificate
k8s.io/kubectl/pkg/util/completion
k8s.io/kubectl/pkg/util/deployment