-
Notifications
You must be signed in to change notification settings - Fork 43
/
processor.go
110 lines (95 loc) · 2.82 KB
/
processor.go
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
package controlplane
import (
"context"
"encoding/json"
"fmt"
"math"
"math/rand"
"os"
"reflect"
"strconv"
"time"
"github.com/envoyproxy/go-control-plane/pkg/cache/types"
"github.com/envoyproxy/go-control-plane/pkg/cache/v3"
"github.com/envoyproxy/go-control-plane/pkg/resource/v3"
"github.com/sirupsen/logrus"
utilcache "k8s.io/apimachinery/pkg/util/cache"
"k8s.io/apimachinery/pkg/util/yaml"
)
type Processor struct {
cache cache.SnapshotCache
logger *logrus.Logger
version int64
expireCache *utilcache.Expiring
}
func NewProcessor(cache cache.SnapshotCache, log *logrus.Logger) *Processor {
return &Processor{
cache: cache,
logger: log,
version: rand.Int63n(1000),
expireCache: utilcache.NewExpiring(),
}
}
func (p *Processor) newVersion() string {
if p.version == math.MaxInt64 {
p.version = 0
}
p.version++
return strconv.FormatInt(p.version, 10)
}
func (p *Processor) ProcessFile(file NotifyMessage) {
configList, err := ParseYaml(file.FilePath)
if err != nil {
p.logger.Errorf("error parsing yaml file: %+v", err)
return
}
for _, config := range configList {
if len(config.Uid) == 0 {
continue
}
lastConfig, ok := p.expireCache.Get(config.Uid)
if ok && reflect.DeepEqual(lastConfig.(*Virtual), config) {
marshal, _ := json.Marshal(config)
p.logger.Debugf("config are same, not needs to update, config: %s", string(marshal))
continue
}
p.logger.Debugf("update config, version %d, config %v", p.version, config)
listeners, clusters, routes, endpoints := config.To()
resources := map[resource.Type][]types.Resource{
resource.ListenerType: listeners, // listeners
resource.RouteType: routes, // routes
resource.ClusterType: clusters, // clusters
resource.EndpointType: endpoints, // endpoints
resource.RuntimeType: {}, // runtimes
resource.SecretType: {}, // secrets
}
var snapshot *cache.Snapshot
snapshot, err = cache.NewSnapshot(p.newVersion(), resources)
if err != nil {
p.logger.Errorf("snapshot inconsistency: %v, err: %v", snapshot, err)
return
}
if err = snapshot.Consistent(); err != nil {
p.logger.Errorf("snapshot inconsistency: %v, err: %v", snapshot, err)
return
}
p.logger.Debugf("will serve snapshot %+v, nodeID: %s", snapshot, config.Uid)
if err = p.cache.SetSnapshot(context.Background(), config.Uid, snapshot); err != nil {
p.logger.Errorf("snapshot error %q for %v", err, snapshot)
p.logger.Fatal(err)
}
p.expireCache.Set(config.Uid, config, time.Minute*5)
}
}
func ParseYaml(file string) ([]*Virtual, error) {
var virtualList = make([]*Virtual, 0)
yamlFile, err := os.ReadFile(file)
if err != nil {
return nil, fmt.Errorf("Error reading YAML file: %s\n", err)
}
err = yaml.Unmarshal(yamlFile, &virtualList)
if err != nil {
return nil, err
}
return virtualList, nil
}