1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
|
package tasks
import (
"fmt"
"os"
"path/filepath"
"strings"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/pb/worker_pb"
)
// TaskLogHandler handles task log requests from admin server
type TaskLogHandler struct {
baseLogDir string
}
// NewTaskLogHandler creates a new task log handler
func NewTaskLogHandler(baseLogDir string) *TaskLogHandler {
if baseLogDir == "" {
baseLogDir = "/tmp/seaweedfs/task_logs"
}
return &TaskLogHandler{
baseLogDir: baseLogDir,
}
}
// HandleLogRequest processes a task log request and returns the response
func (h *TaskLogHandler) HandleLogRequest(request *worker_pb.TaskLogRequest) *worker_pb.TaskLogResponse {
response := &worker_pb.TaskLogResponse{
TaskId: request.TaskId,
WorkerId: request.WorkerId,
Success: false,
}
// Find the task log directory
logDir, err := h.findTaskLogDirectory(request.TaskId)
if err != nil {
response.ErrorMessage = fmt.Sprintf("Task log directory not found: %v", err)
glog.Warningf("Task log request failed for %s: %v", request.TaskId, err)
return response
}
// Read metadata if requested
if request.IncludeMetadata {
metadata, err := h.readTaskMetadata(logDir)
if err != nil {
response.ErrorMessage = fmt.Sprintf("Failed to read task metadata: %v", err)
glog.Warningf("Failed to read metadata for task %s: %v", request.TaskId, err)
return response
}
response.Metadata = metadata
}
// Read log entries
logEntries, err := h.readTaskLogEntries(logDir, request)
if err != nil {
response.ErrorMessage = fmt.Sprintf("Failed to read task logs: %v", err)
glog.Warningf("Failed to read logs for task %s: %v", request.TaskId, err)
return response
}
response.LogEntries = logEntries
response.Success = true
glog.V(1).Infof("Successfully retrieved %d log entries for task %s", len(logEntries), request.TaskId)
return response
}
// findTaskLogDirectory searches for the task log directory by task ID
func (h *TaskLogHandler) findTaskLogDirectory(taskID string) (string, error) {
entries, err := os.ReadDir(h.baseLogDir)
if err != nil {
return "", fmt.Errorf("failed to read base log directory: %w", err)
}
// Look for directories that start with the task ID
for _, entry := range entries {
if entry.IsDir() && strings.HasPrefix(entry.Name(), taskID+"_") {
return filepath.Join(h.baseLogDir, entry.Name()), nil
}
}
return "", fmt.Errorf("task log directory not found for task ID: %s", taskID)
}
// readTaskMetadata reads task metadata from the log directory
func (h *TaskLogHandler) readTaskMetadata(logDir string) (*worker_pb.TaskLogMetadata, error) {
metadata, err := GetTaskLogMetadata(logDir)
if err != nil {
return nil, err
}
// Convert to protobuf metadata
pbMetadata := &worker_pb.TaskLogMetadata{
TaskId: metadata.TaskID,
TaskType: metadata.TaskType,
WorkerId: metadata.WorkerID,
StartTime: metadata.StartTime.Unix(),
Status: metadata.Status,
Progress: float32(metadata.Progress),
VolumeId: metadata.VolumeID,
Server: metadata.Server,
Collection: metadata.Collection,
LogFilePath: metadata.LogFilePath,
CreatedAt: metadata.CreatedAt.Unix(),
CustomData: make(map[string]string),
}
// Set end time and duration if available
if metadata.EndTime != nil {
pbMetadata.EndTime = metadata.EndTime.Unix()
}
if metadata.Duration != nil {
pbMetadata.DurationMs = metadata.Duration.Milliseconds()
}
// Convert custom data
for key, value := range metadata.CustomData {
if strValue, ok := value.(string); ok {
pbMetadata.CustomData[key] = strValue
} else {
pbMetadata.CustomData[key] = fmt.Sprintf("%v", value)
}
}
return pbMetadata, nil
}
// readTaskLogEntries reads and filters log entries based on the request
func (h *TaskLogHandler) readTaskLogEntries(logDir string, request *worker_pb.TaskLogRequest) ([]*worker_pb.TaskLogEntry, error) {
entries, err := ReadTaskLogs(logDir)
if err != nil {
return nil, err
}
// Apply filters
var filteredEntries []TaskLogEntry
for _, entry := range entries {
// Filter by log level
if request.LogLevel != "" && !strings.EqualFold(entry.Level, request.LogLevel) {
continue
}
// Filter by time range
if request.StartTime > 0 && entry.Timestamp.Unix() < request.StartTime {
continue
}
if request.EndTime > 0 && entry.Timestamp.Unix() > request.EndTime {
continue
}
filteredEntries = append(filteredEntries, entry)
}
// Limit entries if requested
if request.MaxEntries > 0 && len(filteredEntries) > int(request.MaxEntries) {
// Take the most recent entries
start := len(filteredEntries) - int(request.MaxEntries)
filteredEntries = filteredEntries[start:]
}
// Convert to protobuf entries
var pbEntries []*worker_pb.TaskLogEntry
for _, entry := range filteredEntries {
pbEntry := &worker_pb.TaskLogEntry{
Timestamp: entry.Timestamp.Unix(),
Level: entry.Level,
Message: entry.Message,
Fields: make(map[string]string),
}
// Set progress if available
if entry.Progress != nil {
pbEntry.Progress = float32(*entry.Progress)
}
// Set status if available
if entry.Status != nil {
pbEntry.Status = *entry.Status
}
// Convert fields
for key, value := range entry.Fields {
if strValue, ok := value.(string); ok {
pbEntry.Fields[key] = strValue
} else {
pbEntry.Fields[key] = fmt.Sprintf("%v", value)
}
}
pbEntries = append(pbEntries, pbEntry)
}
return pbEntries, nil
}
// ListAvailableTaskLogs returns a list of available task log directories
func (h *TaskLogHandler) ListAvailableTaskLogs() ([]string, error) {
entries, err := os.ReadDir(h.baseLogDir)
if err != nil {
return nil, fmt.Errorf("failed to read base log directory: %w", err)
}
var taskDirs []string
for _, entry := range entries {
if entry.IsDir() {
taskDirs = append(taskDirs, entry.Name())
}
}
return taskDirs, nil
}
// CleanupOldLogs removes old task logs beyond the specified limit
func (h *TaskLogHandler) CleanupOldLogs(maxTasks int) error {
config := TaskLoggerConfig{
BaseLogDir: h.baseLogDir,
MaxTasks: maxTasks,
}
// Create a temporary logger to trigger cleanup
tempLogger := &FileTaskLogger{
config: config,
}
tempLogger.cleanupOldLogs()
return nil
}
|
