falcon性能优化之hbs篇

Posted on | In | | 浏览 次

上一篇呢主要是讲了对于judge模块的内存优化，并且由于涉及到了judge与hbs之间的接口修改，因此在hbs这边也要进行少量的修改来进行优化，下面就直接简单介绍下优化思路。

优化思路

在judge哪边每次同步都是根据key来进行同步的，在最初的版本中hbs是把自己的全部数据使用list的方式全部推送给judge，当然啦，judge可以选择自己build成list然后筛选出来自己需要的数据，但是build是个很耗时的过程，同时拿全量数据对于带宽是个很大的消耗，并且judge对于增量更新是强需求，所以hbs这边必须支持依据key来进行同步的功能。

在这种情况下就必须要把原来的关键数据结构（list）改为map了，key是host+metric，在修改完成之后呢，发现hbs的内存上升了3-4G，这个很好理解哈，因为原本的结构只有list没有那个key这占用了很大一部分内存，之后通过比对数据结构意识到，其实呢，对于hbs来说即便是原本的list那种结构其实也是浪费了极大地内存的。为什么呢？

原因就在于，hbs当中存储的模板+策略其实毕竟是少数，但是在应用到节点上的时候就会导致每一个策略对象都会复制一遍。假设我们本来仅仅是一个模板然后里面有100台机器。那么这种情况下在hbs当中这一个模板+十条规则占用的内存就变成了原来的100倍！所以这种情况下其实就应该用指针啦，改用指针之后内存就会极大地减少，这里我贴一下关键数据结构，然后就接着说往下说（因为这是第一版）

1
2
3
4
5
6
type StrategyTpl struct {
    StyPtr *model.Strategy
    TplPtr *model.Template
  //这里不管是模板还是策略我们都用指针，这样就不用把每一个内存都copy一份生成新的对象了
  //所以相同的道理，在judge哪边如果想要进行继续优化修改成这种结构也是ok的，但是投入产出比并不会太高哈
}

修改到这里内存就从12G减小到了比原来最初的8G还要稍微少一丢丢的内存占用，这种情况下judge从hbs同步数据的时间也从最初的3分钟左右变成了几十秒。

下图内存明显分位三段，第一段是使用list时hbs的内存占用，第二段是修改为map之后的内存占用，第三段是修改为指针后的内存占用。

hbsV1

还没完

随着日常的使用呢，经过8个月发现hbs的内存占用会出现极大地波动，最多的时候内存占用会翻倍，发现这个情况之后找了很久都没有找到原因，因为数据结构的优化已经基本达到了极限，这时候突然想起来，将list改为map的时候内存出现了大量的增长，这个巨大的内存波动，很有可能是hbs这个大字典的key导致的，于是进行了验证，将字符串的key经过一层hash变为int64，之后发现内存占用果然下降了，不进内存占用比8个月前list的还小，并且内存波动也从原来的翻倍变成了1G之内的波动。这里注意一下，本质上是使用时间换空间，因此对于cpu的需求会上升。

最初是一个大字典然后使用了一个全局锁，这个其实很浪费性能在judge同步hbs数据的时候是会很影响时间的，因此这次优化主要包括以下几点：

  1. 将host+metric的字符串key经过一层hash映射为唯一的int64
  2. 将大map修改为支持并发锁的字典
  3. 在计算告警规则的时候使用协程并行计算规则，组建字典

下面直接贴代码了

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
import (
cmap "github.com/orcaman/concurrent-mapi" //这里使用了这个map
)

//因为计算key的hash会涉及到频繁的内存分配和释放，因此这里使用pool减少内存压力 :)
var buffPool = sync.Pool{New: func() interface{} { return new(bytes.Buffer) }}

const (
    offset64 = 14695981039346656037
    prime64  = 1099511628211
)

func hashNew() uint64 {
    return offset64
}

func string2uint64(s string) uint64 {
    h := hashNew()
    for i := 0; i < len(s); i++ {
        h ^= uint64(s[i])
        h *= prime64
    }
    return h
}

type StrategyMapC struct {
    M cmap.ConcurrentMap
}
//这里需要给并发锁字典增加一些方法
func (m *StrategyMapC) Set(key uint64, ss []StrategyTpl) {
    m.M.Set(key, ss)
}
func (m *StrategyMapC) Get(key uint64) (ss []StrategyTpl, exist bool) {
    value, ok := m.M.Get(key)
    if ok {
        if v, tok := value.([]StrategyTpl); tok {
            ss = v
            exist = true
        } else {
            exist = false
        }
    } else {
        exist = false
    }
    return

}

func (m *StrategyMapC) Remove(key uint64) {
    m.M.Remove(key)
}
func FilterStrategies(dfs model.DataFilter) map[string][]model.Strategy {
    lock.RLock()
    defer lock.RUnlock()
    m := make(map[string][]model.Strategy)
    for key, _ := range dfs {
        //在根据judge请求的key查找规则的时候也需要做一次转换
        key_i := string2uint64(key)
        if v_list, exist := StrategiesMap.Get(key_i); exist {
            for _, p := range v_list {
                strategy := *(p.StyPtr)
                strategy.Tpl = p.TplPtr
                m[key] = append(m[key], strategy)
            }
        }
    }
    return m
}

func SafeStrategiesResponseInit() {
    // 一个机器ID对应多个模板ID
    hidTids := HostTemplateIds.GetMap()
    sz := len(hidTids)
    if sz == 0 {
        return
    }

    // Judge需要的是hostname，此处要把HostId转换为hostname
    // 查出的hosts，是不处于维护时间内的
    hosts := MonitoredHosts.Get()
    if len(hosts) == 0 {
        // 所有机器都处于维护状态，汗
        return
    }

    tpls := TemplateCache.GetMap()
    if len(tpls) == 0 {
        return
    }

    strategies := Strategies.GetMap()
    if len(strategies) == 0 {
        return
    }

    // 做个索引，给一个tplId，可以很方便的找到对应了哪些Strategy
      var strategy_i int
    var wg, wgk sync.WaitGroup
    //control_c := make(chan struct{}, runtime.NumCPU()) //这里原本是想根据cpu数量控制加载的协程数量
    control_c := make(chan struct{}, 4)
    key_chan := make(chan uint64, runtime.NumCPU()*50)
    wgk.Add(1)
    go func() {
        defer func() { wgk.Done() }()
        for k := range key_chan {
            delete(StrategyKeys, k)
        }
    }()

    for hostId, tplIds := range hidTids {

        h, exists := hosts[hostId]
        if !exists {
            continue
        }

        // 过滤掉指定机房的机器
        hostname := h.Name
        if len(hostname) < 3 {
            continue
        }
              hostnamePrefix := hostname[:3]
        hostnameSuffix := hostname[len(hostname)-3 : len(hostname)]

        // hulk机器的机房标识需要特殊处理下
        if hostnamePrefix == "set" {
            if len(hostname) < 7 {
                continue
            } else {
                hostnamePrefix = hostname[4:7]
            }
        }

        var ok1, ok2 bool
        ignoreIdcMap := g.Config().IgnoreIdcMap
        _, ok1 = ignoreIdcMap[hostnamePrefix]
        _, ok2 = ignoreIdcMap[hostnameSuffix]
        if ok1 || ok2 {
            continue
        }

        // 计算当前host配置了哪些监控策略
       这里使用channel来控制并发度
        control_c <- struct{}{}
        wg.Add(1)
        go func(hostid int, tplids []int) {
            defer func() { <-control_c; wg.Done() }()
            allStrategies := CalcInheritStrategies(tpls, tplids)
                      if len(allStrategies) <= 0 {
                return
            }
            var available_strategiesTpl []StrategyTpl
            // 检查策略列表里面是否存在被禁用的策略
            disabled_strategy_ids, exists := HostStrategyMap.GetStrategyIds(hostid)
            if !exists {
                available_strategiesTpl = allStrategies
            } else {
                for _, valStrategyTpl := range allStrategies {
                    // 判断当前策略id是否在禁用策略列表里面
                    if Slice_int_contains(disabled_strategy_ids, valStrategyTpl.StyPtr.Id) {
                        continue
                    }
                    available_strategiesTpl = append(available_strategiesTpl, valStrategyTpl)
                }
            }
            available_num := len(available_strategiesTpl)
            if available_num <= 0 {
                return
            }
            strategy_i += available_num
                      host_metric_strategy := make(map[uint64][]StrategyTpl, 50)
            ///////////////////////////////////////////////
            buf := buffPool.Get().(*bytes.Buffer)
            for _, str_tpl_ptr := range available_strategiesTpl {

                buf.Reset()
                buf.Write([]byte(h.Name))
                buf.Write([]byte("/"))
                buf.Write([]byte(str_tpl_ptr.StyPtr.Metric))
                key := buf.String()
								//这里将字符串映射为int64
                key_i := string2uint64(key)
                if _, exist := host_metric_strategy[key_i]; exist {
                    host_metric_strategy[key_i] = append(host_metric_strategy[key_i], str_tpl_ptr)
                } else {
                    host_metric_strategy[key_i] = []StrategyTpl{str_tpl_ptr}
                }
            }
            buffPool.Put(buf)
            for key_i, ss := range host_metric_strategy {
                StrategiesMap.Set(key_i, ss)
                key_chan <- key_i

            }
        }(hostId, tplIds)
    }
    wg.Wait()
      close(key_chan)
    wgk.Wait()
    for key, _ := range StrategyKeys {
        StrategiesMap.Remove(key)
    }
 

    return
}

下图是修改完的hbs实例和未修改实例的内存比对图，可以看到，不管是内存占用还是内存波动优化效果都十分明显

hbsV2

PS

以上对于hbs的优化方法不管是string转int还是使用并发锁对于graph同样适用，有需求的同学可以实践一下

转载请注明来源链接 http://just4fun.im/2020/03/01/falcon-optimize-hbs/ 尊重知识,谢谢:)