62 lines
1.5 KiB
Go
62 lines
1.5 KiB
Go
package checkpoint
|
||
|
||
import (
|
||
"context"
|
||
|
||
"github.com/redis/go-redis/v9"
|
||
)
|
||
|
||
// Redis 键设计（与 database-design.md 6 节对齐）：
|
||
// - 日志文件 offset：hms:agent:logoffset:{file}
|
||
// - 主机已确认 seq：hms:agent:ackedseq:{host_id}
|
||
const (
|
||
keyLogOffset = "hms:agent:logoffset:"
|
||
keyAckedSeq = "hms:agent:ackedseq:"
|
||
)
|
||
|
||
// RedisStore 使用 Redis 记录检查点，实现 Agent 重启/漂移后的断点续传。
|
||
type RedisStore struct {
|
||
cli *redis.Client
|
||
}
|
||
|
||
// NewRedisStore 解析 Redis URL 并创建客户端。
|
||
func NewRedisStore(url string) (*RedisStore, error) {
|
||
opt, err := redis.ParseURL(url)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return &RedisStore{cli: redis.NewClient(opt)}, nil
|
||
}
|
||
|
||
func (s *RedisStore) GetLogOffset(file string) (int64, error) {
|
||
v, err := s.cli.Get(context.Background(), keyLogOffset+file).Int64()
|
||
if err == redis.Nil {
|
||
return 0, nil
|
||
}
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
return v, nil
|
||
}
|
||
|
||
func (s *RedisStore) SetLogOffset(file string, offset int64) error {
|
||
return s.cli.Set(context.Background(), keyLogOffset+file, offset, 0).Err()
|
||
}
|
||
|
||
func (s *RedisStore) GetAckedSeq(hostID string) (int64, error) {
|
||
v, err := s.cli.Get(context.Background(), keyAckedSeq+hostID).Int64()
|
||
if err == redis.Nil {
|
||
return 0, nil
|
||
}
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
return v, nil
|
||
}
|
||
|
||
func (s *RedisStore) SetAckedSeq(hostID string, seq int64) error {
|
||
return s.cli.Set(context.Background(), keyAckedSeq+hostID, seq, 0).Err()
|
||
}
|
||
|
||
func (s *RedisStore) Close() error { return s.cli.Close() }
|