inject_util.go

Documentation: github.com/linkerd/linkerd2/cli/cmd

     1  package cmd
     2  
     3  import (
     4  	"bufio"
     5  	"bytes"
     6  	"errors"
     7  	"fmt"
     8  	"io"
     9  	"net/http"
    10  	"net/url"
    11  	"os"
    12  	"path/filepath"
    13  	"strings"
    14  
    15  	"github.com/linkerd/linkerd2/pkg/inject"
    16  	corev1 "k8s.io/api/core/v1"
    17  	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    18  	"k8s.io/apimachinery/pkg/runtime"
    19  	yamlDecoder "k8s.io/apimachinery/pkg/util/yaml"
    20  	"sigs.k8s.io/yaml"
    21  )
    22  
    23  type resourceTransformer interface {
    24  	transform([]byte) ([]byte, []inject.Report, error)
    25  	generateReport([]inject.Report, io.Writer)
    26  }
    27  
    28  // Returns the integer representation of os.Exit code; 0 on success and 1 on failure.
    29  func transformInput(inputs []io.Reader, errWriter, outWriter io.Writer, rt resourceTransformer, format string) int {
    30  	postInjectBuf := &bytes.Buffer{}
    31  	reportBuf := &bytes.Buffer{}
    32  
    33  	for _, input := range inputs {
    34  		errs := processYAML(input, postInjectBuf, reportBuf, rt, format)
    35  		if len(errs) > 0 {
    36  			fmt.Fprintf(errWriter, "Error transforming resources:\n%v", concatErrors(errs, "\n"))
    37  			return 1
    38  		}
    39  
    40  		_, err := io.Copy(outWriter, postInjectBuf)
    41  
    42  		// print error report after yaml output, for better visibility
    43  		io.Copy(errWriter, reportBuf)
    44  
    45  		if err != nil {
    46  			fmt.Fprintf(errWriter, "Error printing YAML: %v\n", err)
    47  			return 1
    48  		}
    49  	}
    50  	return 0
    51  }
    52  
    53  // processYAML takes an input stream of YAML, outputting injected/uninjected YAML to out.
    54  func processYAML(in io.Reader, out io.Writer, report io.Writer, rt resourceTransformer, format string) []error {
    55  	reader := yamlDecoder.NewYAMLReader(bufio.NewReaderSize(in, 4096))
    56  
    57  	reports := []inject.Report{}
    58  
    59  	errs := []error{}
    60  
    61  	// Iterate over all YAML objects in the input
    62  	for {
    63  		// Read a single YAML object
    64  		bytes, err := reader.Read()
    65  		if err != nil {
    66  			if errors.Is(err, io.EOF) {
    67  				break
    68  			}
    69  			return []error{err}
    70  		}
    71  
    72  		var result []byte
    73  		var irs []inject.Report
    74  
    75  		isList, err := kindIsList(bytes)
    76  		if err != nil {
    77  			return []error{err}
    78  		}
    79  		if isList {
    80  			result, irs, err = processList(bytes, rt)
    81  		} else {
    82  			result, irs, err = rt.transform(bytes)
    83  		}
    84  		if err != nil {
    85  			errs = append(errs, err)
    86  		}
    87  		reports = append(reports, irs...)
    88  
    89  		// If the format is set to json, we need to convert the yaml to json
    90  		if format == "json" {
    91  			result, err = yaml.YAMLToJSON(result)
    92  			if err != nil {
    93  				errs = append(errs, err)
    94  			}
    95  		} else if format == "yaml" {
    96  			// result is already in yaml format: noop.
    97  		} else {
    98  			errs = append(errs, fmt.Errorf("unsupported format %s", format))
    99  		}
   100  
   101  		if len(errs) == 0 {
   102  			out.Write(result)
   103  			if format == "yaml" {
   104  				out.Write([]byte("---\n"))
   105  			}
   106  			if format == "json" {
   107  				out.Write([]byte("\n"))
   108  			}
   109  		}
   110  	}
   111  
   112  	rt.generateReport(reports, report)
   113  
   114  	return errs
   115  }
   116  
   117  func kindIsList(bytes []byte) (bool, error) {
   118  	var meta metav1.TypeMeta
   119  	if err := yaml.Unmarshal(bytes, &meta); err != nil {
   120  		return false, err
   121  	}
   122  	return meta.Kind == "List", nil
   123  }
   124  
   125  func processList(bytes []byte, rt resourceTransformer) ([]byte, []inject.Report, error) {
   126  	var sourceList corev1.List
   127  	if err := yaml.Unmarshal(bytes, &sourceList); err != nil {
   128  		return nil, nil, err
   129  	}
   130  
   131  	reports := []inject.Report{}
   132  	items := []runtime.RawExtension{}
   133  
   134  	for _, item := range sourceList.Items {
   135  		result, irs, err := rt.transform(item.Raw)
   136  		if err != nil {
   137  			return nil, nil, err
   138  		}
   139  
   140  		// At this point, we have yaml. The kubernetes internal representation is
   141  		// json. Because we're building a list from RawExtensions, the yaml needs
   142  		// to be converted to json.
   143  		injected, err := yaml.YAMLToJSON(result)
   144  		if err != nil {
   145  			return nil, nil, err
   146  		}
   147  
   148  		items = append(items, runtime.RawExtension{Raw: injected})
   149  		reports = append(reports, irs...)
   150  	}
   151  
   152  	sourceList.Items = items
   153  	result, err := yaml.Marshal(sourceList)
   154  	if err != nil {
   155  		return nil, nil, err
   156  	}
   157  	return result, reports, nil
   158  }
   159  
   160  // Read all the resource files found in path into a slice of readers.
   161  // path can be either a file, directory or stdin.
   162  func read(path string) ([]io.Reader, error) {
   163  	if path == "-" {
   164  		return []io.Reader{os.Stdin}, nil
   165  	}
   166  
   167  	if url, ok := toURL(path); ok {
   168  		if strings.ToLower(url.Scheme) != "https" {
   169  			return nil, fmt.Errorf("only HTTPS URLs are allowed")
   170  		}
   171  		resp, err := http.Get(url.String())
   172  		if err != nil {
   173  			return nil, err
   174  		}
   175  		defer resp.Body.Close()
   176  
   177  		if resp.StatusCode != http.StatusOK {
   178  			return nil, fmt.Errorf("unable to read URL %q, server reported %s, status code=%d", path, resp.Status, resp.StatusCode)
   179  		}
   180  
   181  		// Save to a buffer, so that response can be closed here
   182  		buf := new(bytes.Buffer)
   183  		_, err = buf.ReadFrom(resp.Body)
   184  		if err != nil {
   185  			return nil, err
   186  		}
   187  
   188  		return []io.Reader{buf}, nil
   189  	}
   190  
   191  	return walk(path)
   192  }
   193  
   194  // checks if the given string is a valid URL
   195  func toURL(path string) (*url.URL, bool) {
   196  	u, err := url.ParseRequestURI(path)
   197  	if err == nil && u.Host != "" && u.Scheme != "" {
   198  		return u, true
   199  	}
   200  
   201  	return nil, false
   202  }
   203  
   204  // walk walks the file tree rooted at path. path may be a file or a directory.
   205  // Creates a reader for each file found.
   206  func walk(path string) ([]io.Reader, error) {
   207  	p := filepath.Clean(path)
   208  	stat, err := os.Stat(p)
   209  	if err != nil {
   210  		return nil, err
   211  	}
   212  
   213  	if !stat.IsDir() {
   214  		file, err := os.Open(p)
   215  		if err != nil {
   216  			return nil, err
   217  		}
   218  
   219  		return []io.Reader{file}, nil
   220  	}
   221  
   222  	var in []io.Reader
   223  	werr := filepath.Walk(p, func(path string, info os.FileInfo, err error) error {
   224  		if err != nil {
   225  			return err
   226  		}
   227  
   228  		if info.IsDir() {
   229  			return nil
   230  		}
   231  
   232  		file, err := os.Open(filepath.Clean(path))
   233  		if err != nil {
   234  			return err
   235  		}
   236  
   237  		in = append(in, file)
   238  		return nil
   239  	})
   240  
   241  	if werr != nil {
   242  		return nil, werr
   243  	}
   244  
   245  	return in, nil
   246  }
   247  
   248  // a helper function to concatenate the items in a []error
   249  // into a single error
   250  func concatErrors(errs []error, delimiter string) error {
   251  	message, errs := errs[0].Error(), errs[1:] // pop the first element of the errs
   252  	// this is done so that the first error message is not prefixed by the delimiter
   253  
   254  	for _, err := range errs {
   255  		message = fmt.Sprintf("%s%s%s", message, delimiter, err.Error())
   256  	}
   257  	return errors.New(message)
   258  }
   259
View as plain text