Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/opencensus/trace/time_event.py: 50%

1# Copyright 2017, OpenCensus Authors 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14 

15from opencensus.common import utils 

16 

17 

18class Type(object): 

19 """ 

20 Indicates whether the message was sent or received. 

21 

22 Attributes: 

23 TYPE_UNSPECIFIED (int): Unknown event type. 

24 SENT (int): Indicates a sent message. 

25 RECEIVED (int): Indicates a received message. 

26 """ 

27 TYPE_UNSPECIFIED = 0 

28 SENT = 1 

29 RECEIVED = 2 

30 

31 

32class Annotation(object): 

33 """Text annotation with a set of attributes. 

34 

35 :type timestamp: :class:`~datetime.datetime` 

36 :param timestamp: The timestamp indicating the time the event occurred. 

37 

38 :type description: str 

39 :param description: A user-supplied message describing the event. 

40 The maximum length for the description is 256 bytes. 

41 

42 :type attributes: :class:`~opencensus.trace.attributes.Attributes` 

43 :param attributes: A set of attributes on the annotation. 

44 You can have up to 4 attributes per Annotation. 

45 """ 

46 def __init__(self, timestamp, description, attributes=None): 

47 self.timestamp = utils.to_iso_str(timestamp) 

48 self.description = description 

49 self.attributes = attributes 

50 

51 def format_annotation_json(self): 

52 annotation_json = {} 

53 annotation_json['description'] = utils.get_truncatable_str( 

54 self.description) 

55 

56 if self.attributes is not None: 

57 annotation_json['attributes'] = self.attributes.\ 

58 format_attributes_json() 

59 

60 return annotation_json 

61 

62 

63class MessageEvent(object): 

64 """An event describing a message sent/received between Spans. 

65 

66 :type timestamp: :class:`~datetime.datetime` 

67 :param timestamp: The timestamp indicating the time the event occurred. 

68 

69 :type type: Enum of :class: `~opencensus.trace.time_event.Type` 

70 :param type: Indicates whether the message was sent or received. 

71 

72 :type id: str (int64 format) 

73 :param id: An identifier for the MessageEvent's message that can be used 

74 to match SENT and RECEIVED MessageEvents. It is recommended to 

75 be unique within a Span. 

76 

77 :type uncompressed_size_bytes: str (int64 format) 

78 :param uncompressed_size_bytes: The number of uncompressed bytes sent or 

79 received. 

80 

81 :type compressed_size_bytes: str (int64 format) 

82 :param compressed_size_bytes: The number of compressed bytes sent or 

83 received. If missing assumed to be the same 

84 size as uncompressed. 

85 

86 """ 

87 def __init__(self, timestamp, id, type=None, uncompressed_size_bytes=None, 

88 compressed_size_bytes=None): 

89 self.timestamp = utils.to_iso_str(timestamp) 

90 

91 if type is None: 

92 type = Type.TYPE_UNSPECIFIED 

93 

94 if compressed_size_bytes is None and \ 

95 uncompressed_size_bytes is not None: 

96 compressed_size_bytes = uncompressed_size_bytes 

97 

98 self.id = id 

99 self.type = type 

100 self.uncompressed_size_bytes = uncompressed_size_bytes 

101 self.compressed_size_bytes = compressed_size_bytes 

102 

103 def format_message_event_json(self): 

104 message_event_json = {} 

105 

106 message_event_json['id'] = self.id 

107 message_event_json['type'] = self.type 

108 

109 if self.uncompressed_size_bytes is not None: 

110 message_event_json[ 

111 'uncompressed_size_bytes'] = self.uncompressed_size_bytes 

112 

113 if self.compressed_size_bytes is not None: 

114 message_event_json[ 

115 'compressed_size_bytes'] = self.compressed_size_bytes 

116 

117 return message_event_json