seaweedfs/weed/admin/dash/admin_data.go

package dash

import (
	"context"
	"net"
	"net/http"
	"sort"
	"time"

	"github.com/seaweedfs/seaweedfs/weed/cluster"
	"github.com/seaweedfs/seaweedfs/weed/glog"
	"github.com/seaweedfs/seaweedfs/weed/iam"
	"github.com/seaweedfs/seaweedfs/weed/pb"
	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
)

// Access key status constants
const (
	AccessKeyStatusActive   = iam.AccessKeyStatusActive
	AccessKeyStatusInactive = iam.AccessKeyStatusInactive
)

type AdminData struct {
	Username          string              `json:"username"`
	TotalVolumes      int                 `json:"total_volumes"`
	TotalChunks       int64               `json:"total_chunks"`
	TotalSize         int64               `json:"total_size"`
	VolumeSizeLimitMB uint64              `json:"volume_size_limit_mb"`
	MasterNodes       []MasterNode        `json:"master_nodes"`
	VolumeServers     []VolumeServer      `json:"volume_servers"`
	FilerNodes        []FilerNode         `json:"filer_nodes"`
	MessageBrokers    []MessageBrokerNode `json:"message_brokers"`
	S3Nodes           []S3Node            `json:"s3_nodes"`
	DataCenters       []DataCenter        `json:"datacenters"`
	LastUpdated       time.Time           `json:"last_updated"`

	// EC shard totals for dashboard
	TotalEcVolumes int `json:"total_ec_volumes"` // Total number of EC volumes across all servers
	TotalEcShards  int `json:"total_ec_shards"`  // Total number of EC shards across all servers

	// TotalMountClients is the number of connected FUSE/VFS mount clients across filers.
	TotalMountClients int `json:"total_mount_clients"`

	// Trends holds at-a-glance sparklines built from the admin's own recent
	// cluster snapshots (no Prometheus required).
	Trends DashboardTrends `json:"trends"`

	// TierStats breaks volumes and EC shards down by storage tier: local
	// disk types plus one entry per remote storage holding tiered volumes.
	TierStats []TierStats `json:"tier_stats"`

	// VersionSummary reports whether all cluster nodes run the same binary
	// version, so the dashboard can warn about partially upgraded clusters.
	VersionSummary ClusterVersionSummary `json:"version_summary"`
}

// Object Store Users management structures
type ObjectStoreUser struct {
	Username    string   `json:"username"`
	Email       string   `json:"email"`
	AccessKey   string   `json:"access_key"`
	SecretKey   string   `json:"secret_key"`
	Permissions []string `json:"permissions"`
	PolicyNames []string `json:"policy_names"`
	IsStatic    bool     `json:"is_static"` // loaded from static config file, not editable
}

type ObjectStoreUsersData struct {
	Username         string            `json:"username"`
	Users            []ObjectStoreUser `json:"users"`
	TotalUsers       int               `json:"total_users"`
	HasAnonymousUser bool              `json:"has_anonymous_user"`
	LastUpdated      time.Time         `json:"last_updated"`
}

// User management request structures
type CreateUserRequest struct {
	Username    string   `json:"username" binding:"required"`
	Email       string   `json:"email"`
	Actions     []string `json:"actions"`
	GenerateKey bool     `json:"generate_key"`
	PolicyNames []string `json:"policy_names"`
}

type UpdateUserRequest struct {
	Email       string   `json:"email"`
	Actions     []string `json:"actions"`
	PolicyNames []string `json:"policy_names"`
}

type UpdateUserPoliciesRequest struct {
	Actions []string `json:"actions" binding:"required"`
}

type AccessKeyInfo struct {
	AccessKey string    `json:"access_key"`
	SecretKey string    `json:"secret_key"`
	Status    string    `json:"status"`
	CreatedAt time.Time `json:"created_at"`
}

type CreateAccessKeyRequest struct {
	AccessKey string `json:"access_key"`
	SecretKey string `json:"secret_key"`
}

type UpdateAccessKeyStatusRequest struct {
	Status string `json:"status" binding:"required"`
}

type UserDetails struct {
	Username    string          `json:"username"`
	Email       string          `json:"email"`
	Actions     []string        `json:"actions"`
	PolicyNames []string        `json:"policy_names"`
	AccessKeys  []AccessKeyInfo `json:"access_keys"`
	Groups      []string        `json:"groups"`
}

// RoleReadOnly is the session role assigned to read-only (view-only) admin
// accounts. It matches the value stored by HandleLogin when the read-only
// credentials are used.
const RoleReadOnly = "readonly"

// IsReadOnlyRole reports whether the given admin session role grants only
// view-only access. Any other role (admin, or the empty role used when auth
// is disabled) is treated as non-read-only.
func IsReadOnlyRole(role string) bool {
	return role == RoleReadOnly
}

// RedactSecretKey clears the plaintext S3 secret key from an object-store
// user record. The access key (a public identifier) is retained so the
// identity can still be listed; only the reusable secret is removed.
func (u *ObjectStoreUser) RedactSecretKey() {
	u.SecretKey = ""
}

// RedactSecretKeys clears the plaintext S3 secret keys from a user's access
// key records. Access key identifiers are retained so the set of keys remains
// visible; only the reusable secrets are removed.
func (d *UserDetails) RedactSecretKeys() {
	for i := range d.AccessKeys {
		d.AccessKeys[i].SecretKey = ""
	}
}

type FilerNode struct {
	Address     string    `json:"address"`
	DataCenter  string    `json:"datacenter"`
	Rack        string    `json:"rack"`
	Version     string    `json:"version"`
	LastUpdated time.Time `json:"last_updated"`
}

type MessageBrokerNode struct {
	Address     string    `json:"address"`
	DataCenter  string    `json:"datacenter"`
	Rack        string    `json:"rack"`
	Version     string    `json:"version"`
	LastUpdated time.Time `json:"last_updated"`
}

type S3Node struct {
	Address     string    `json:"address"`
	DataCenter  string    `json:"datacenter"`
	Version     string    `json:"version"`
	LastUpdated time.Time `json:"last_updated"`
}

// GetAdminData retrieves admin data as a struct (for reuse by both JSON and HTML handlers)
func (s *AdminServer) GetAdminData(username string) (AdminData, error) {
	if username == "" {
		username = "admin"
	}

	// Get cluster topology
	topology, err := s.GetClusterTopology()
	if err != nil {
		glog.Errorf("Failed to get cluster topology: %v", err)
		return AdminData{}, err
	}

	// Get volume servers data with EC shard information
	volumeServersData, err := s.GetClusterVolumeServers()
	if err != nil {
		glog.Errorf("Failed to get cluster volume servers: %v", err)
		return AdminData{}, err
	}

	// Get master nodes status
	masterNodes := s.getMasterNodesStatus()

	// Get filer nodes status
	filerNodes := s.getFilerNodesStatus()

	// Get message broker nodes status
	messageBrokers := s.getMessageBrokerNodesStatus()

	// Get S3 nodes status
	s3Nodes := s.getS3NodesStatus()

	// Get volume size limit from master configuration
	var volumeSizeLimitMB uint64 = 30000 // Default to 30GB
	err = s.WithMasterClient(func(client master_pb.SeaweedClient) error {
		resp, err := client.GetMasterConfiguration(context.Background(), &master_pb.GetMasterConfigurationRequest{})
		if err != nil {
			return err
		}
		volumeSizeLimitMB = uint64(resp.VolumeSizeLimitMB)
		return nil
	})
	if err != nil {
		glog.Warningf("Failed to get volume size limit from master: %v", err)
		// Keep default value on error
	}

	// Calculate EC shard totals
	var totalEcVolumes, totalEcShards int
	ecVolumeSet := make(map[uint32]bool) // To avoid counting the same EC volume multiple times

	for _, vs := range volumeServersData.VolumeServers {
		totalEcShards += vs.EcShards
		// Count unique EC volumes across all servers
		for _, ecInfo := range vs.EcShardDetails {
			ecVolumeSet[ecInfo.VolumeID] = true
		}
	}
	totalEcVolumes = len(ecVolumeSet)

	// Count connected FUSE/VFS mount clients (best-effort; don't fail the dashboard)
	totalMountClients := 0
	if mountData, mountErr := s.GetMountClients(); mountErr == nil {
		totalMountClients = mountData.TotalMountClients
	}

	// Prepare admin data
	adminData := AdminData{
		Username:          username,
		TotalVolumes:      topology.TotalVolumes,
		TotalChunks:       topology.TotalChunks,
		TotalSize:         topology.TotalSize,
		VolumeSizeLimitMB: volumeSizeLimitMB,
		MasterNodes:       masterNodes,
		VolumeServers:     volumeServersData.VolumeServers,
		FilerNodes:        filerNodes,
		MessageBrokers:    messageBrokers,
		S3Nodes:           s3Nodes,
		DataCenters:       topology.DataCenters,
		LastUpdated:       topology.UpdatedAt,
		TotalEcVolumes:    totalEcVolumes,
		TotalEcShards:     totalEcShards,
		TotalMountClients: totalMountClients,
		Trends:            s.GetDashboardTrends(),
		TierStats:         topology.TierStats,
		VersionSummary:    buildClusterVersionSummary(volumeServersData.VolumeServers, filerNodes, s3Nodes, messageBrokers),
	}

	return adminData, nil
}

// ShowAdmin displays the main admin page (now uses GetAdminData)
func (s *AdminServer) ShowAdmin(w http.ResponseWriter, r *http.Request) {
	username := UsernameFromContext(r.Context())

	adminData, err := s.GetAdminData(username)
	if err != nil {
		writeJSONError(w, http.StatusInternalServerError, "Failed to get admin data: "+err.Error())
		return
	}

	// Return JSON for API calls
	writeJSON(w, http.StatusOK, adminData)
}

// ShowOverview displays cluster overview
func (s *AdminServer) ShowOverview(w http.ResponseWriter, r *http.Request) {
	topology, err := s.GetClusterTopology()
	if err != nil {
		writeJSONError(w, http.StatusInternalServerError, err.Error())
		return
	}

	writeJSON(w, http.StatusOK, topology)
}

// getMasterNodesStatus returns the full set of master nodes in the cluster.
// It prefers the authoritative raft membership (RaftListClusterServers) and
// falls back to the currently-connected master if the raft call fails, so the
// dashboard never shows an empty list.
func (s *AdminServer) getMasterNodesStatus() []MasterNode {
	masterMap := make(map[string]MasterNode)
	raftReturnedEmpty := false

	err := s.WithMasterClient(func(client master_pb.SeaweedClient) error {
		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
		defer cancel()
		resp, err := client.RaftListClusterServers(ctx, &master_pb.RaftListClusterServersRequest{})
		if err != nil {
			return err
		}
		raftReturnedEmpty = len(resp.ClusterServers) == 0
		for _, server := range resp.ClusterServers {
			// Skip malformed raft addresses instead of letting an
			// unconvertible value into masterMap.
			if _, _, splitErr := net.SplitHostPort(server.Address); splitErr != nil {
				glog.Warningf("skip master with invalid raft address %q: %v", server.Address, splitErr)
				continue
			}
			httpAddress := pb.GrpcAddressToServerAddress(server.Address)
			if httpAddress == "" {
				glog.Warningf("skip master with invalid raft address %q", server.Address)
				continue
			}
			masterMap[httpAddress] = MasterNode{
				Address:  httpAddress,
				IsLeader: server.IsLeader,
			}
		}
		return nil
	})

	if err != nil {
		currentMaster := s.masterClient.GetMaster(context.Background())
		glog.Errorf("Failed to list raft cluster masters from %s: %v", currentMaster, err)
	}

	if len(masterMap) == 0 {
		currentMaster := s.masterClient.GetMaster(context.Background())
		if currentMaster != "" {
			addr := pb.ServerAddress(currentMaster).ToHttpAddress()
			// A successful empty raft response means raft is not initialized
			// (standalone/non-raft cluster); the only master IS the leader.
			// A failed RPC or a nonempty response whose entries were all
			// rejected must not claim leadership.
			masterMap[addr] = MasterNode{
				Address:  addr,
				IsLeader: raftReturnedEmpty,
			}
		}
	}

	masterNodes := make([]MasterNode, 0, len(masterMap))
	for _, m := range masterMap {
		masterNodes = append(masterNodes, m)
	}
	sort.Slice(masterNodes, func(i, j int) bool {
		return masterNodes[i].Address < masterNodes[j].Address
	})
	return masterNodes
}

// getFilerNodesStatus checks status of all filer nodes using master's ListClusterNodes
func (s *AdminServer) getFilerNodesStatus() []FilerNode {
	var filerNodes []FilerNode

	// Get filer nodes from master using ListClusterNodes
	err := s.WithMasterClient(func(client master_pb.SeaweedClient) error {
		resp, err := client.ListClusterNodes(context.Background(), s.listClusterNodesRequest(cluster.FilerType))
		if err != nil {
			return err
		}

		// Process each filer node
		for _, node := range resp.ClusterNodes {
			filerNodes = append(filerNodes, FilerNode{
				Address:     pb.ServerAddress(node.Address).ToHttpAddress(),
				DataCenter:  node.DataCenter,
				Rack:        node.Rack,
				Version:     node.Version,
				LastUpdated: time.Now(),
			})
		}

		return nil
	})

	if err != nil {
		currentMaster := s.masterClient.GetMaster(context.Background())
		glog.Errorf("Failed to get filer nodes from master %s: %v", currentMaster, err)
		// Return empty list if we can't get filer info from master
		return []FilerNode{}
	}

	// Sort filer nodes by address for consistent ordering on page refresh
	sort.Slice(filerNodes, func(i, j int) bool {
		return filerNodes[i].Address < filerNodes[j].Address
	})

	return filerNodes
}

// getMessageBrokerNodesStatus checks status of all message broker nodes using master's ListClusterNodes
func (s *AdminServer) getMessageBrokerNodesStatus() []MessageBrokerNode {
	var messageBrokers []MessageBrokerNode

	// Get message broker nodes from master using ListClusterNodes
	err := s.WithMasterClient(func(client master_pb.SeaweedClient) error {
		resp, err := client.ListClusterNodes(context.Background(), s.listClusterNodesRequest(cluster.BrokerType))
		if err != nil {
			return err
		}

		// Process each message broker node
		for _, node := range resp.ClusterNodes {
			messageBrokers = append(messageBrokers, MessageBrokerNode{
				Address:     node.Address,
				DataCenter:  node.DataCenter,
				Rack:        node.Rack,
				Version:     node.Version,
				LastUpdated: time.Now(),
			})
		}

		return nil
	})

	if err != nil {
		currentMaster := s.masterClient.GetMaster(context.Background())
		glog.Errorf("Failed to get message broker nodes from master %s: %v", currentMaster, err)
		// Return empty list if we can't get broker info from master
		return []MessageBrokerNode{}
	}

	// Sort message broker nodes by address for consistent ordering on page refresh
	sort.Slice(messageBrokers, func(i, j int) bool {
		return messageBrokers[i].Address < messageBrokers[j].Address
	})

	return messageBrokers
}

// getS3NodesStatus checks status of all S3 nodes using master's ListClusterNodes
func (s *AdminServer) getS3NodesStatus() []S3Node {
	var s3Nodes []S3Node

	// Get S3 nodes from master using ListClusterNodes
	err := s.WithMasterClient(func(client master_pb.SeaweedClient) error {
		resp, err := client.ListClusterNodes(context.Background(), s.listClusterNodesRequest(cluster.S3Type))
		if err != nil {
			return err
		}

		// Process each S3 node
		for _, node := range resp.ClusterNodes {
			s3Nodes = append(s3Nodes, S3Node{
				Address:     pb.ServerAddress(node.Address).ToHttpAddress(),
				DataCenter:  node.DataCenter,
				Version:     node.Version,
				LastUpdated: time.Now(),
			})
		}

		return nil
	})

	if err != nil {
		currentMaster := s.masterClient.GetMaster(context.Background())
		glog.Errorf("Failed to get S3 nodes from master %s: %v", currentMaster, err)
		// Return empty list if we can't get S3 info from master
		return []S3Node{}
	}

	// Sort S3 nodes by address for consistent ordering on page refresh
	sort.Slice(s3Nodes, func(i, j int) bool {
		return s3Nodes[i].Address < s3Nodes[j].Address
	})

	return s3Nodes
}