pilot/pkg/xds/debug.go (849 lines of code) (raw):
// Copyright Istio 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 xds
import (
"encoding/json"
"fmt"
"html/template"
"net"
"net/http"
"net/http/pprof"
"sort"
"strings"
"time"
)
import (
adminapi "github.com/envoyproxy/go-control-plane/envoy/admin/v3"
core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
wasm "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/wasm/v3"
tls "github.com/envoyproxy/go-control-plane/envoy/extensions/transport_sockets/tls/v3"
discovery "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
"google.golang.org/protobuf/proto"
any "google.golang.org/protobuf/types/known/anypb"
istiolog "istio.io/pkg/log"
)
import (
"github.com/apache/dubbo-go-pixiu/pilot/pkg/config/kube/crd"
"github.com/apache/dubbo-go-pixiu/pilot/pkg/features"
"github.com/apache/dubbo-go-pixiu/pilot/pkg/model"
"github.com/apache/dubbo-go-pixiu/pilot/pkg/networking/util"
"github.com/apache/dubbo-go-pixiu/pilot/pkg/serviceregistry"
"github.com/apache/dubbo-go-pixiu/pilot/pkg/serviceregistry/aggregate"
"github.com/apache/dubbo-go-pixiu/pilot/pkg/serviceregistry/memory"
"github.com/apache/dubbo-go-pixiu/pilot/pkg/serviceregistry/provider"
v3 "github.com/apache/dubbo-go-pixiu/pilot/pkg/xds/v3"
"github.com/apache/dubbo-go-pixiu/pkg/config"
"github.com/apache/dubbo-go-pixiu/pkg/config/schema/collection"
"github.com/apache/dubbo-go-pixiu/pkg/config/xds"
"github.com/apache/dubbo-go-pixiu/pkg/network"
"github.com/apache/dubbo-go-pixiu/pkg/util/protomarshal"
)
var indexTmpl = template.Must(template.New("index").Parse(`<html>
<head>
<title>Pilot Debug Console</title>
</head>
<style>
#endpoints {
font-family: "Trebuchet MS", Arial, Helvetica, sans-serif;
border-collapse: collapse;
}
#endpoints td, #endpoints th {
border: 1px solid #ddd;
padding: 8px;
}
#endpoints tr:nth-child(even){background-color: #f2f2f2;}
#endpoints tr:hover {background-color: #ddd;}
#endpoints th {
padding-top: 12px;
padding-bottom: 12px;
text-align: left;
background-color: black;
color: white;
}
</style>
<body>
<br/>
<table id="endpoints">
<tr><th>Endpoint</th><th>Description</th></tr>
{{range .}}
<tr>
<td><a href='{{.Href}}'>{{.Name}}</a></td><td>{{.Help}}</td>
</tr>
{{end}}
</table>
<br/>
</body>
</html>
`))
// AdsClient defines the data that is displayed on "/adsz" endpoint.
type AdsClient struct {
ConnectionID string `json:"connectionId"`
ConnectedAt time.Time `json:"connectedAt"`
PeerAddress string `json:"address"`
Metadata *model.NodeMetadata `json:"metadata,omitempty"`
Watches map[string][]string `json:"watches,omitempty"`
}
// AdsClients is collection of AdsClient connected to this Istiod.
type AdsClients struct {
Total int `json:"totalClients"`
Connected []AdsClient `json:"clients,omitempty"`
}
// SyncStatus is the synchronization status between Pilot and a given Envoy
type SyncStatus struct {
ClusterID string `json:"cluster_id,omitempty"`
ProxyID string `json:"proxy,omitempty"`
ProxyVersion string `json:"proxy_version,omitempty"`
IstioVersion string `json:"istio_version,omitempty"`
ClusterSent string `json:"cluster_sent,omitempty"`
ClusterAcked string `json:"cluster_acked,omitempty"`
ListenerSent string `json:"listener_sent,omitempty"`
ListenerAcked string `json:"listener_acked,omitempty"`
RouteSent string `json:"route_sent,omitempty"`
RouteAcked string `json:"route_acked,omitempty"`
EndpointSent string `json:"endpoint_sent,omitempty"`
EndpointAcked string `json:"endpoint_acked,omitempty"`
ExtensionConfigSent string `json:"extensionconfig_sent,omitempty"`
ExtensionConfigAcked string `json:"extensionconfig_acked,omitempty"`
}
// SyncedVersions shows what resourceVersion of a given resource has been acked by Envoy.
type SyncedVersions struct {
ProxyID string `json:"proxy,omitempty"`
ClusterVersion string `json:"cluster_acked,omitempty"`
ListenerVersion string `json:"listener_acked,omitempty"`
RouteVersion string `json:"route_acked,omitempty"`
}
// InitDebug initializes the debug handlers and adds a debug in-memory registry.
func (s *DiscoveryServer) InitDebug(mux *http.ServeMux, sctl *aggregate.Controller, enableProfiling bool,
fetchWebhook func() map[string]string) {
// For debugging and load testing v2 we add an memory registry.
s.MemRegistry = memory.NewServiceDiscovery()
s.MemRegistry.EDSUpdater = s
s.MemRegistry.ClusterID = "v2-debug"
sctl.AddRegistry(serviceregistry.Simple{
ClusterID: "v2-debug",
ProviderID: provider.Mock,
ServiceDiscovery: s.MemRegistry,
Controller: s.MemRegistry.Controller,
})
internalMux := http.NewServeMux()
s.AddDebugHandlers(mux, internalMux, enableProfiling, fetchWebhook)
debugGen, ok := (s.Generators[TypeDebug]).(*DebugGen)
if ok {
debugGen.DebugMux = internalMux
}
}
func (s *DiscoveryServer) AddDebugHandlers(mux, internalMux *http.ServeMux, enableProfiling bool, webhook func() map[string]string) {
// Debug handlers on HTTP ports are added for backward compatibility.
// They will be exposed on XDS-over-TLS in future releases.
if !features.EnableDebugOnHTTP {
return
}
if enableProfiling {
s.addDebugHandler(mux, internalMux, "/debug/pprof/", "Displays pprof index", pprof.Index)
s.addDebugHandler(mux, internalMux, "/debug/pprof/cmdline", "The command line invocation of the current program", pprof.Cmdline)
s.addDebugHandler(mux, internalMux, "/debug/pprof/profile", "CPU profile", pprof.Profile)
s.addDebugHandler(mux, internalMux, "/debug/pprof/symbol", "Symbol looks up the program counters listed in the request", pprof.Symbol)
s.addDebugHandler(mux, internalMux, "/debug/pprof/trace", "A trace of execution of the current program.", pprof.Trace)
}
mux.HandleFunc("/debug", s.Debug)
if features.EnableUnsafeAdminEndpoints {
s.addDebugHandler(mux, internalMux, "/debug/force_disconnect", "Disconnects a proxy from this Pilot", s.forceDisconnect)
}
s.addDebugHandler(mux, internalMux, "/debug/ecdsz", "Status and debug interface for ECDS", s.ecdsz)
s.addDebugHandler(mux, internalMux, "/debug/edsz", "Status and debug interface for EDS", s.Edsz)
s.addDebugHandler(mux, internalMux, "/debug/ndsz", "Status and debug interface for NDS", s.ndsz)
s.addDebugHandler(mux, internalMux, "/debug/adsz", "Status and debug interface for ADS", s.adsz)
s.addDebugHandler(mux, internalMux, "/debug/adsz?push=true", "Initiates push of the current state to all connected endpoints", s.adsz)
s.addDebugHandler(mux, internalMux, "/debug/syncz", "Synchronization status of all Envoys connected to this Pilot instance", s.Syncz)
s.addDebugHandler(mux, internalMux, "/debug/config_distribution", "Version status of all Envoys connected to this Pilot instance", s.distributedVersions)
s.addDebugHandler(mux, internalMux, "/debug/registryz", "Debug support for registry", s.registryz)
s.addDebugHandler(mux, internalMux, "/debug/endpointz", "Debug support for endpoints", s.endpointz)
s.addDebugHandler(mux, internalMux, "/debug/endpointShardz", "Info about the endpoint shards", s.endpointShardz)
s.addDebugHandler(mux, internalMux, "/debug/cachez", "Info about the internal XDS caches", s.cachez)
s.addDebugHandler(mux, internalMux, "/debug/cachez?sizes=true", "Info about the size of the internal XDS caches", s.cachez)
s.addDebugHandler(mux, internalMux, "/debug/cachez?clear=true", "Clear the XDS caches", s.cachez)
s.addDebugHandler(mux, internalMux, "/debug/configz", "Debug support for config", s.configz)
s.addDebugHandler(mux, internalMux, "/debug/sidecarz", "Debug sidecar scope for a proxy", s.sidecarz)
s.addDebugHandler(mux, internalMux, "/debug/resourcesz", "Debug support for watched resources", s.resourcez)
s.addDebugHandler(mux, internalMux, "/debug/instancesz", "Debug support for service instances", s.instancesz)
s.addDebugHandler(mux, internalMux, "/debug/authorizationz", "Internal authorization policies", s.authorizationz)
s.addDebugHandler(mux, internalMux, "/debug/telemetryz", "Debug Telemetry configuration", s.telemetryz)
s.addDebugHandler(mux, internalMux, "/debug/config_dump", "ConfigDump in the form of the Envoy admin config dump API for passed in proxyID", s.ConfigDump)
s.addDebugHandler(mux, internalMux, "/debug/push_status", "Last PushContext Details", s.pushStatusHandler)
s.addDebugHandler(mux, internalMux, "/debug/pushcontext", "Debug support for current push context", s.pushContextHandler)
s.addDebugHandler(mux, internalMux, "/debug/connections", "Info about the connected XDS clients", s.connectionsHandler)
s.addDebugHandler(mux, internalMux, "/debug/inject", "Active inject template", s.injectTemplateHandler(webhook))
s.addDebugHandler(mux, internalMux, "/debug/mesh", "Active mesh config", s.meshHandler)
s.addDebugHandler(mux, internalMux, "/debug/clusterz", "List remote clusters where istiod reads endpoints", s.clusterz)
s.addDebugHandler(mux, internalMux, "/debug/networkz", "List cross-network gateways", s.networkz)
s.addDebugHandler(mux, internalMux, "/debug/mcsz", "List information about Kubernetes MCS services", s.mcsz)
s.addDebugHandler(mux, internalMux, "/debug/list", "List all supported debug commands in json", s.List)
}
func (s *DiscoveryServer) addDebugHandler(mux *http.ServeMux, internalMux *http.ServeMux,
path string, help string, handler func(http.ResponseWriter, *http.Request)) {
s.debugHandlers[path] = help
// Add handler without auth. This mux is never exposed on an HTTP server and only used internally
if internalMux != nil {
internalMux.HandleFunc(path, handler)
}
// Add handler with auth; this is expose on an HTTP server
mux.HandleFunc(path, s.allowAuthenticatedOrLocalhost(http.HandlerFunc(handler)))
}
func (s *DiscoveryServer) allowAuthenticatedOrLocalhost(next http.Handler) http.HandlerFunc {
return func(w http.ResponseWriter, req *http.Request) {
// Request is from localhost, no need to authenticate
if isRequestFromLocalhost(req) {
next.ServeHTTP(w, req)
return
}
// Authenticate request with the same method as XDS
authFailMsgs := make([]string, 0)
var ids []string
for _, authn := range s.Authenticators {
u, err := authn.AuthenticateRequest(req)
// If one authenticator passes, return
if u != nil && u.Identities != nil && err == nil {
ids = u.Identities
break
}
authFailMsgs = append(authFailMsgs, fmt.Sprintf("Authenticator %s: %v", authn.AuthenticatorType(), err))
}
if ids == nil {
istiolog.Errorf("Failed to authenticate %s %v", req.URL, authFailMsgs)
// Not including detailed info in the response, XDS doesn't either (returns a generic "authentication failure).
w.WriteHeader(http.StatusUnauthorized)
return
}
// TODO: Check that the identity contains dubbo-system namespace, else block or restrict to only info that
// is visible to the authenticated SA. Will require changes in docs and istioctl too.
next.ServeHTTP(w, req)
}
}
func isRequestFromLocalhost(r *http.Request) bool {
ip, _, err := net.SplitHostPort(r.RemoteAddr)
if err != nil {
return false
}
userIP := net.ParseIP(ip)
return userIP.IsLoopback()
}
// Syncz dumps the synchronization status of all Envoys connected to this Pilot instance
func (s *DiscoveryServer) Syncz(w http.ResponseWriter, _ *http.Request) {
syncz := make([]SyncStatus, 0)
for _, con := range s.Clients() {
node := con.proxy
if node != nil {
syncz = append(syncz, SyncStatus{
ProxyID: node.ID,
ClusterID: node.Metadata.ClusterID.String(),
IstioVersion: node.Metadata.IstioVersion,
ClusterSent: con.NonceSent(v3.ClusterType),
ClusterAcked: con.NonceAcked(v3.ClusterType),
ListenerSent: con.NonceSent(v3.ListenerType),
ListenerAcked: con.NonceAcked(v3.ListenerType),
RouteSent: con.NonceSent(v3.RouteType),
RouteAcked: con.NonceAcked(v3.RouteType),
EndpointSent: con.NonceSent(v3.EndpointType),
EndpointAcked: con.NonceAcked(v3.EndpointType),
ExtensionConfigSent: con.NonceSent(v3.ExtensionConfigurationType),
ExtensionConfigAcked: con.NonceAcked(v3.ExtensionConfigurationType),
})
}
}
writeJSON(w, syncz)
}
// registryz providees debug support for registry - adding and listing model items.
// Can be combined with the push debug interface to reproduce changes.
func (s *DiscoveryServer) registryz(w http.ResponseWriter, req *http.Request) {
all := s.Env.ServiceDiscovery.Services()
writeJSON(w, all)
}
// Dumps info about the endpoint shards, tracked using the new direct interface.
// Legacy registry provides are synced to the new data structure as well, during
// the full push.
func (s *DiscoveryServer) endpointShardz(w http.ResponseWriter, req *http.Request) {
w.Header().Add("Content-Type", "application/json")
out, _ := json.MarshalIndent(s.EndpointIndex.Shardz(), " ", " ")
_, _ = w.Write(out)
}
func (s *DiscoveryServer) cachez(w http.ResponseWriter, req *http.Request) {
if err := req.ParseForm(); err != nil {
w.WriteHeader(http.StatusBadRequest)
_, _ = w.Write([]byte("Failed to parse request\n"))
return
}
if req.Form.Get("clear") != "" {
s.Cache.ClearAll()
_, _ = w.Write([]byte("Cache cleared\n"))
return
}
if req.Form.Get("sizes") != "" {
snapshot := s.Cache.Snapshot()
res := make(map[string]string, len(snapshot))
totalSize := 0
for _, resource := range snapshot {
if resource == nil {
continue
}
resourceType := resource.Resource.TypeUrl
sz := len(resource.Resource.GetValue())
res[resourceType] += util.ByteCount(sz)
totalSize += sz
}
res["total"] = util.ByteCount(totalSize)
writeJSON(w, res)
return
}
snapshot := s.Cache.Snapshot()
resources := make(map[string][]string, len(snapshot)) // Key is typeUrl and value is resource names.
for key, resource := range snapshot {
if resource == nil {
continue
}
resourceType := resource.Resource.TypeUrl
resources[resourceType] = append(resources[resourceType], resource.Name+"/"+key)
}
writeJSON(w, resources)
}
type endpointzResponse struct {
Service string `json:"svc"`
Endpoints []*model.ServiceInstance `json:"ep"`
}
// Endpoint debugging
func (s *DiscoveryServer) endpointz(w http.ResponseWriter, req *http.Request) {
if _, f := req.URL.Query()["brief"]; f {
svc := s.Env.ServiceDiscovery.Services()
for _, ss := range svc {
for _, p := range ss.Ports {
all := s.Env.ServiceDiscovery.InstancesByPort(ss, p.Port, nil)
for _, svc := range all {
_, _ = fmt.Fprintf(w, "%s:%s %s:%d %v %s\n", ss.Hostname,
p.Name, svc.Endpoint.Address, svc.Endpoint.EndpointPort, svc.Endpoint.Labels,
svc.Endpoint.ServiceAccount)
}
}
}
return
}
svc := s.Env.ServiceDiscovery.Services()
resp := make([]endpointzResponse, 0)
for _, ss := range svc {
for _, p := range ss.Ports {
all := s.Env.ServiceDiscovery.InstancesByPort(ss, p.Port, nil)
resp = append(resp, endpointzResponse{
Service: fmt.Sprintf("%s:%s", ss.Hostname, p.Name),
Endpoints: all,
})
}
}
writeJSON(w, resp)
}
func (s *DiscoveryServer) distributedVersions(w http.ResponseWriter, req *http.Request) {
if !features.EnableDistributionTracking {
w.WriteHeader(http.StatusConflict)
_, _ = fmt.Fprint(w, "Pilot Version tracking is disabled. Please set the "+
"PILOT_ENABLE_CONFIG_DISTRIBUTION_TRACKING environment variable to true to enable.")
return
}
if resourceID := req.URL.Query().Get("resource"); resourceID != "" {
proxyNamespace := req.URL.Query().Get("proxy_namespace")
knownVersions := make(map[string]string)
var results []SyncedVersions
for _, con := range s.Clients() {
// wrap this in independent scope so that panic's don't bypass Unlock...
con.proxy.RLock()
if con.proxy != nil && (proxyNamespace == "" || proxyNamespace == con.proxy.ConfigNamespace) {
// read nonces from our statusreporter to allow for skipped nonces, etc.
results = append(results, SyncedVersions{
ProxyID: con.proxy.ID,
ClusterVersion: s.getResourceVersion(s.StatusReporter.QueryLastNonce(con.conID, v3.ClusterType),
resourceID, knownVersions),
ListenerVersion: s.getResourceVersion(s.StatusReporter.QueryLastNonce(con.conID, v3.ListenerType),
resourceID, knownVersions),
RouteVersion: s.getResourceVersion(s.StatusReporter.QueryLastNonce(con.conID, v3.RouteType),
resourceID, knownVersions),
})
}
con.proxy.RUnlock()
}
writeJSON(w, results)
} else {
w.WriteHeader(http.StatusUnprocessableEntity)
_, _ = fmt.Fprintf(w, "querystring parameter 'resource' is required\n")
}
}
// VersionLen is the Config Version and is only used as the nonce prefix, but we can reconstruct
// it because is is a b64 encoding of a 64 bit array, which will always be 12 chars in length.
// len = ceil(bitlength/(2^6))+1
const VersionLen = 12
func (s *DiscoveryServer) getResourceVersion(nonce, key string, cache map[string]string) string {
if len(nonce) < VersionLen {
return ""
}
configVersion := nonce[:VersionLen]
result, ok := cache[configVersion]
if !ok {
lookupResult, err := s.Env.GetLedger().GetPreviousValue(configVersion, key)
if err != nil {
istiolog.Errorf("Unable to retrieve resource %s at version %s: %v", key, configVersion, err)
lookupResult = ""
}
// update the cache even on an error, because errors will not resolve themselves, and we don't want to
// repeat the same error for many s.adsClients.
cache[configVersion] = lookupResult
return lookupResult
}
return result
}
// kubernetesConfig wraps a config.Config with a custom marshaling method that matches a Kubernetes
// object structure.
type kubernetesConfig struct {
config.Config
}
func (k kubernetesConfig) MarshalJSON() ([]byte, error) {
cfg, err := crd.ConvertConfig(k.Config)
if err != nil {
return nil, err
}
return json.Marshal(cfg)
}
// Config debugging.
func (s *DiscoveryServer) configz(w http.ResponseWriter, req *http.Request) {
configs := make([]kubernetesConfig, 0)
s.Env.ConfigStore.Schemas().ForEach(func(schema collection.Schema) bool {
cfg, _ := s.Env.ConfigStore.List(schema.Resource().GroupVersionKind(), "")
for _, c := range cfg {
configs = append(configs, kubernetesConfig{c})
}
return false
})
writeJSON(w, configs)
}
// SidecarScope debugging
func (s *DiscoveryServer) sidecarz(w http.ResponseWriter, req *http.Request) {
proxyID, con := s.getDebugConnection(req)
if con == nil {
s.errorHandler(w, proxyID, con)
return
}
writeJSON(w, con.proxy.SidecarScope)
}
// Resource debugging.
func (s *DiscoveryServer) resourcez(w http.ResponseWriter, _ *http.Request) {
schemas := make([]config.GroupVersionKind, 0)
s.Env.Schemas().ForEach(func(schema collection.Schema) bool {
schemas = append(schemas, schema.Resource().GroupVersionKind())
return false
})
writeJSON(w, schemas)
}
// AuthorizationDebug holds debug information for authorization policy.
type AuthorizationDebug struct {
AuthorizationPolicies *model.AuthorizationPolicies `json:"authorization_policies"`
}
// authorizationz dumps the internal authorization policies.
func (s *DiscoveryServer) authorizationz(w http.ResponseWriter, req *http.Request) {
info := AuthorizationDebug{
AuthorizationPolicies: s.globalPushContext().AuthzPolicies,
}
writeJSON(w, info)
}
// AuthorizationDebug holds debug information for authorization policy.
type TelemetryDebug struct {
Telemetries *model.Telemetries `json:"telemetries"`
}
func (s *DiscoveryServer) telemetryz(w http.ResponseWriter, req *http.Request) {
info := TelemetryDebug{
Telemetries: s.globalPushContext().Telemetry,
}
writeJSON(w, info)
}
// connectionsHandler implements interface for displaying current connections.
// It is mapped to /debug/connections.
func (s *DiscoveryServer) connectionsHandler(w http.ResponseWriter, req *http.Request) {
adsClients := &AdsClients{}
connections := s.Clients()
adsClients.Total = len(connections)
for _, c := range connections {
adsClient := AdsClient{
ConnectionID: c.conID,
ConnectedAt: c.connectedAt,
PeerAddress: c.peerAddr,
}
adsClients.Connected = append(adsClients.Connected, adsClient)
}
writeJSON(w, adsClients)
}
// adsz implements a status and debug interface for ADS.
// It is mapped to /debug/adsz
func (s *DiscoveryServer) adsz(w http.ResponseWriter, req *http.Request) {
if s.handlePushRequest(w, req) {
return
}
proxyID, con := s.getDebugConnection(req)
if proxyID != "" && con == nil {
// We can't guarantee the Pilot we are connected to has a connection to the proxy we requested
// There isn't a great way around this, but for debugging purposes its suitable to have the caller retry.
w.WriteHeader(http.StatusNotFound)
_, _ = w.Write([]byte("Proxy not connected to this Pilot instance. It may be connected to another instance.\n"))
return
}
var connections []*Connection
if con != nil {
connections = []*Connection{con}
} else {
connections = s.Clients()
}
adsClients := &AdsClients{}
adsClients.Total = len(connections)
for _, c := range connections {
adsClient := AdsClient{
ConnectionID: c.conID,
ConnectedAt: c.connectedAt,
PeerAddress: c.peerAddr,
Metadata: c.proxy.Metadata,
Watches: map[string][]string{},
}
c.proxy.RLock()
for k, wr := range c.proxy.WatchedResources {
r := wr.ResourceNames
if r == nil {
r = []string{}
}
adsClient.Watches[k] = r
}
c.proxy.RUnlock()
adsClients.Connected = append(adsClients.Connected, adsClient)
}
sort.Slice(adsClients.Connected, func(i, j int) bool {
return adsClients.Connected[i].ConnectionID < adsClients.Connected[j].ConnectionID
})
writeJSON(w, adsClients)
}
// ecdsz implements a status and debug interface for ECDS.
// It is mapped to /debug/ecdsz
func (s *DiscoveryServer) ecdsz(w http.ResponseWriter, req *http.Request) {
if s.handlePushRequest(w, req) {
return
}
proxyID, con := s.getDebugConnection(req)
if con == nil {
s.errorHandler(w, proxyID, con)
return
}
if s.Generators[v3.ExtensionConfigurationType] != nil {
r, ok := con.proxy.WatchedResources[v3.ExtensionConfigurationType]
if !ok {
w.WriteHeader(http.StatusNotFound)
_, _ = w.Write([]byte(fmt.Sprintf("no watched ExtensionConfigurationType found, proxyID: %s\n", proxyID)))
return
}
resource, _, _ := s.Generators[v3.ExtensionConfigurationType].Generate(con.proxy, r, &model.PushRequest{
Full: true,
Push: con.proxy.LastPushContext,
})
if len(resource) == 0 {
w.WriteHeader(http.StatusNotFound)
_, _ = w.Write([]byte(fmt.Sprintf("ExtensionConfigurationType not found, proxyID: %s\n", proxyID)))
return
}
wasmCfgs := make([]interface{}, 0, len(resource))
for _, rr := range resource {
if w, err := unmarshalToWasm(rr); err != nil {
istiolog.Warnf("failed to unmarshal wasm: %v", err)
} else {
wasmCfgs = append(wasmCfgs, w)
}
}
writeJSON(w, wasmCfgs)
}
}
func unmarshalToWasm(r *discovery.Resource) (interface{}, error) {
tce := &core.TypedExtensionConfig{}
if err := r.GetResource().UnmarshalTo(tce); err != nil {
return nil, err
}
switch tce.TypedConfig.TypeUrl {
case xds.WasmHTTPFilterType:
w := &wasm.Wasm{}
if err := tce.TypedConfig.UnmarshalTo(w); err != nil {
return nil, err
}
// Redact Wasm secret env variable.
vmenvs := w.GetConfig().GetVmConfig().EnvironmentVariables
if vmenvs != nil {
if _, found := vmenvs.KeyValues[model.WasmSecretEnv]; found {
vmenvs.KeyValues[model.WasmSecretEnv] = "<Redacted>"
}
}
return w, nil
}
return tce, nil
}
// ConfigDump returns information in the form of the Envoy admin API config dump for the specified proxy
// The dump will only contain dynamic listeners/clusters/routes and can be used to compare what an Envoy instance
// should look like according to Pilot vs what it currently does look like.
func (s *DiscoveryServer) ConfigDump(w http.ResponseWriter, req *http.Request) {
proxyID, con := s.getDebugConnection(req)
if con == nil {
s.errorHandler(w, proxyID, con)
return
}
dump, err := s.configDump(con)
if err != nil {
handleHTTPError(w, err)
return
}
writeJSON(w, dump)
}
// configDump converts the connection internal state into an Envoy Admin API config dump proto
// It is used in debugging to create a consistent object for comparison between Envoy and Pilot outputs
func (s *DiscoveryServer) configDump(conn *Connection) (*adminapi.ConfigDump, error) {
dynamicActiveClusters := make([]*adminapi.ClustersConfigDump_DynamicCluster, 0)
req := &model.PushRequest{Push: conn.proxy.LastPushContext, Start: time.Now()}
clusters, _ := s.ConfigGenerator.BuildClusters(conn.proxy, req)
for _, cs := range clusters {
dynamicActiveClusters = append(dynamicActiveClusters, &adminapi.ClustersConfigDump_DynamicCluster{Cluster: cs.Resource})
}
clustersAny, err := util.MessageToAnyWithError(&adminapi.ClustersConfigDump{
VersionInfo: versionInfo(),
DynamicActiveClusters: dynamicActiveClusters,
})
if err != nil {
return nil, err
}
dynamicActiveListeners := make([]*adminapi.ListenersConfigDump_DynamicListener, 0)
listeners := s.ConfigGenerator.BuildListeners(conn.proxy, req.Push)
for _, cs := range listeners {
listener, err := any.New(cs)
if err != nil {
return nil, err
}
dynamicActiveListeners = append(dynamicActiveListeners, &adminapi.ListenersConfigDump_DynamicListener{
Name: cs.Name,
ActiveState: &adminapi.ListenersConfigDump_DynamicListenerState{Listener: listener},
})
}
listenersAny, err := util.MessageToAnyWithError(&adminapi.ListenersConfigDump{
VersionInfo: versionInfo(),
DynamicListeners: dynamicActiveListeners,
})
if err != nil {
return nil, err
}
routes, _ := s.ConfigGenerator.BuildHTTPRoutes(conn.proxy, req, conn.Routes())
routeConfigAny := util.MessageToAny(&adminapi.RoutesConfigDump{})
if len(routes) > 0 {
dynamicRouteConfig := make([]*adminapi.RoutesConfigDump_DynamicRouteConfig, 0)
for _, rs := range routes {
dynamicRouteConfig = append(dynamicRouteConfig, &adminapi.RoutesConfigDump_DynamicRouteConfig{RouteConfig: rs.Resource})
}
routeConfigAny, err = util.MessageToAnyWithError(&adminapi.RoutesConfigDump{DynamicRouteConfigs: dynamicRouteConfig})
if err != nil {
return nil, err
}
}
secretsDump := &adminapi.SecretsConfigDump{}
if s.Generators[v3.SecretType] != nil {
secrets, _, _ := s.Generators[v3.SecretType].Generate(conn.proxy, conn.Watched(v3.SecretType), nil)
if len(secrets) > 0 {
for _, secretAny := range secrets {
secret := &tls.Secret{}
if err := secretAny.GetResource().UnmarshalTo(secret); err != nil {
istiolog.Warnf("failed to unmarshal secret: %v", err)
}
if secret.GetTlsCertificate() != nil {
secret.GetTlsCertificate().PrivateKey = &core.DataSource{
Specifier: &core.DataSource_InlineBytes{
InlineBytes: []byte("[redacted]"),
},
}
}
secretsDump.DynamicActiveSecrets = append(secretsDump.DynamicActiveSecrets, &adminapi.SecretsConfigDump_DynamicSecret{
Name: secret.Name,
Secret: util.MessageToAny(secret),
})
}
}
}
bootstrapAny := util.MessageToAny(&adminapi.BootstrapConfigDump{})
scopedRoutesAny := util.MessageToAny(&adminapi.ScopedRoutesConfigDump{})
// The config dump must have all configs with connections specified in
// https://www.envoyproxy.io/docs/envoy/latest/api-v2/admin/v2alpha/config_dump.proto
configDump := &adminapi.ConfigDump{
Configs: []*any.Any{
bootstrapAny,
clustersAny, listenersAny,
scopedRoutesAny,
routeConfigAny,
util.MessageToAny(secretsDump),
},
}
return configDump, nil
}
// injectTemplateHandler dumps the injection template
// Replaces dumping the template at startup.
func (s *DiscoveryServer) injectTemplateHandler(webhook func() map[string]string) func(http.ResponseWriter, *http.Request) {
return func(w http.ResponseWriter, req *http.Request) {
// TODO: we should split the inject template into smaller modules (separate one for dump core, etc),
// and allow pods to select which patches will be selected. When this happen, this should return
// all inject templates or take a param to select one.
if webhook == nil {
w.WriteHeader(http.StatusNotFound)
return
}
writeJSON(w, webhook())
}
}
// meshHandler dumps the mesh config
func (s *DiscoveryServer) meshHandler(w http.ResponseWriter, r *http.Request) {
writeJSON(w, s.Env.Mesh())
}
// pushStatusHandler dumps the last PushContext
func (s *DiscoveryServer) pushStatusHandler(w http.ResponseWriter, req *http.Request) {
model.LastPushMutex.Lock()
defer model.LastPushMutex.Unlock()
if model.LastPushStatus == nil {
return
}
out, err := model.LastPushStatus.StatusJSON()
if err != nil {
handleHTTPError(w, err)
return
}
w.Header().Add("Content-Type", "application/json")
_, _ = w.Write(out)
}
// PushContextDebug holds debug information for push context.
type PushContextDebug struct {
AuthorizationPolicies *model.AuthorizationPolicies
NetworkGateways map[network.ID][]model.NetworkGateway
}
// pushContextHandler dumps the current PushContext
func (s *DiscoveryServer) pushContextHandler(w http.ResponseWriter, _ *http.Request) {
push := PushContextDebug{
AuthorizationPolicies: s.globalPushContext().AuthzPolicies,
NetworkGateways: s.globalPushContext().NetworkManager().GatewaysByNetwork(),
}
writeJSON(w, push)
}
// Debug lists all the supported debug endpoints.
func (s *DiscoveryServer) Debug(w http.ResponseWriter, req *http.Request) {
type debugEndpoint struct {
Name string
Href string
Help string
}
var deps []debugEndpoint
for k, v := range s.debugHandlers {
deps = append(deps, debugEndpoint{
Name: k,
Href: k,
Help: v,
})
}
sort.Slice(deps, func(i, j int) bool {
return deps[i].Name < deps[j].Name
})
if err := indexTmpl.Execute(w, deps); err != nil {
istiolog.Errorf("Error in rendering index template %v", err)
w.WriteHeader(http.StatusInternalServerError)
}
}
// List all the supported debug commands in json.
func (s *DiscoveryServer) List(w http.ResponseWriter, req *http.Request) {
var cmdNames []string
for k := range s.debugHandlers {
key := strings.Replace(k, "/debug/", "", -1)
// exclude current list command
if key == "list" {
continue
}
// can not support pprof commands
if strings.Contains(key, "pprof") {
continue
}
cmdNames = append(cmdNames, key)
}
sort.Strings(cmdNames)
writeJSON(w, cmdNames)
}
// ndsz implements a status and debug interface for NDS.
// It is mapped to /debug/ndsz on the monitor port (15014).
func (s *DiscoveryServer) ndsz(w http.ResponseWriter, req *http.Request) {
if s.handlePushRequest(w, req) {
return
}
proxyID, con := s.getDebugConnection(req)
if con == nil {
s.errorHandler(w, proxyID, con)
return
}
if !con.proxy.Metadata.DNSCapture {
w.WriteHeader(http.StatusBadRequest)
_, _ = w.Write([]byte("DNS capture is not enabled in the proxy\n"))
return
}
if s.Generators[v3.NameTableType] != nil {
nds, _, _ := s.Generators[v3.NameTableType].Generate(con.proxy, nil, &model.PushRequest{
Push: con.proxy.LastPushContext,
})
if len(nds) == 0 {
return
}
writeJSON(w, nds[0])
}
}
// Edsz implements a status and debug interface for EDS.
// It is mapped to /debug/edsz on the monitor port (15014).
func (s *DiscoveryServer) Edsz(w http.ResponseWriter, req *http.Request) {
if s.handlePushRequest(w, req) {
return
}
proxyID, con := s.getDebugConnection(req)
if con == nil {
s.errorHandler(w, proxyID, con)
return
}
clusters := con.Clusters()
eps := make([]jsonMarshalProto, 0, len(clusters))
for _, clusterName := range clusters {
eps = append(eps, jsonMarshalProto{s.generateEndpoints(NewEndpointBuilder(clusterName, con.proxy, con.proxy.LastPushContext))})
}
writeJSON(w, eps)
}
func (s *DiscoveryServer) forceDisconnect(w http.ResponseWriter, req *http.Request) {
proxyID, con := s.getDebugConnection(req)
if con == nil {
s.errorHandler(w, proxyID, con)
return
}
con.Stop()
_, _ = w.Write([]byte("OK"))
}
func (s *DiscoveryServer) getProxyConnection(proxyID string) *Connection {
for _, con := range s.Clients() {
if strings.Contains(con.conID, proxyID) {
return con
}
}
return nil
}
func (s *DiscoveryServer) instancesz(w http.ResponseWriter, req *http.Request) {
instances := map[string][]*model.ServiceInstance{}
for _, con := range s.Clients() {
con.proxy.RLock()
if con.proxy != nil {
instances[con.proxy.ID] = con.proxy.ServiceInstances
}
con.proxy.RUnlock()
}
writeJSON(w, instances)
}
func (s *DiscoveryServer) networkz(w http.ResponseWriter, _ *http.Request) {
writeJSON(w, s.Env.NetworkManager.AllGateways())
}
func (s *DiscoveryServer) mcsz(w http.ResponseWriter, _ *http.Request) {
svcs := sortMCSServices(s.Env.MCSServices())
writeJSON(w, svcs)
}
func sortMCSServices(svcs []model.MCSServiceInfo) []model.MCSServiceInfo {
sort.Slice(svcs, func(i, j int) bool {
if strings.Compare(svcs[i].Cluster.String(), svcs[j].Cluster.String()) < 0 {
return true
}
if strings.Compare(svcs[i].Namespace, svcs[j].Namespace) < 0 {
return true
}
return strings.Compare(svcs[i].Name, svcs[j].Name) < 0
})
return svcs
}
func (s *DiscoveryServer) clusterz(w http.ResponseWriter, _ *http.Request) {
if s.ListRemoteClusters == nil {
w.WriteHeader(400)
return
}
writeJSON(w, s.ListRemoteClusters())
}
// handlePushRequest handles a ?push=true query param and triggers a push.
// A boolean response is returned to indicate if the caller should continue
func (s *DiscoveryServer) handlePushRequest(w http.ResponseWriter, req *http.Request) bool {
if err := req.ParseForm(); err != nil {
w.WriteHeader(http.StatusBadRequest)
_, _ = w.Write([]byte("Failed to parse request\n"))
return true
}
if req.Form.Get("push") != "" {
AdsPushAll(s)
_, _ = fmt.Fprintf(w, "Pushed to %d servers\n", s.adsClientCount())
return true
}
return false
}
// getDebugConnection fetches the Connection requested by proxyID
func (s *DiscoveryServer) getDebugConnection(req *http.Request) (string, *Connection) {
if proxyID := req.URL.Query().Get("proxyID"); proxyID != "" {
return proxyID, s.getProxyConnection(proxyID)
}
return "", nil
}
func (s *DiscoveryServer) errorHandler(w http.ResponseWriter, proxyID string, con *Connection) {
if proxyID == "" {
w.WriteHeader(http.StatusBadRequest)
_, _ = w.Write([]byte("You must provide a proxyID in the query string\n"))
return
}
if con == nil {
// We can't guarantee the Pilot we are connected to has a connection to the proxy we requested
// There isn't a great way around this, but for debugging purposes its suitable to have the caller retry.
w.WriteHeader(http.StatusNotFound)
_, _ = w.Write([]byte("Proxy not connected to this Pilot instance. It may be connected to another instance.\n"))
return
}
}
// jsonMarshalProto wraps a proto.Message so it can be marshaled with the standard encoding/json library
type jsonMarshalProto struct {
proto.Message
}
func (p jsonMarshalProto) MarshalJSON() ([]byte, error) {
return protomarshal.Marshal(p.Message)
}
// writeJSON writes a json payload, handling content type, marshaling, and errors
func writeJSON(w http.ResponseWriter, obj interface{}) {
w.Header().Set("Content-Type", "application/json")
b, err := config.ToJSON(obj)
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
_, _ = w.Write([]byte(err.Error()))
return
}
_, err = w.Write(b)
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
}
}
// handleHTTPError writes an error message to the response
func handleHTTPError(w http.ResponseWriter, err error) {
w.WriteHeader(http.StatusInternalServerError)
_, _ = w.Write([]byte(err.Error()))
}